工程化爬虫开发规范

在入门教程中，我们学会了如何写一个能用的爬虫。但在实际生产环境中，一个"能用"的爬虫远远不够。本章将带你从工程化的角度重新审视爬虫开发，让你的爬虫代码更加健壮、可维护、易于调试。

学习目标：掌握日志系统、配置管理、异常处理和项目结构设计，为后续进阶学习打下坚实基础。

为什么需要工程化

当你的爬虫从简单的脚本演变为需要长期运行、多人协作的项目时，以下问题会逐渐浮现：

• 调试困难：程序出错了，但不知道错在哪里
• 配置混乱：API密钥、数据库连接等硬编码在代码中
• 脆弱性高：网络波动就导致整个程序崩溃
• 可读性差：几个月后自己都看不懂当初写的代码

工程化开发规范正是为了解决这些问题而存在的。本章我们将重点学习：

1. 使用 loguru 构建专业的日志系统
2. 使用 pydantic-settings 进行配置管理
3. 实现统一的异常处理和重试机制
4. 项目目录结构的最佳实践

工程化核心模块

---

日志系统设计

为什么需要专业的日志系统

在入门教程中，我们大量使用 print() 来输出信息。这种方式在调试时很方便，但存在以下问题：

• 无法区分信息的重要程度（调试信息和错误信息混在一起）
• 无法持久化保存日志
• 无法在生产环境中动态调整日志级别
• 无法记录时间、文件位置等上下文信息

loguru 是一个优秀的 Python 日志库，它用极简的 API 解决了以上所有问题。

loguru 基本使用

首先安装 loguru：

pip install loguru

最简单的使用方式：

from loguru import logger

logger.debug("这是一条调试信息")
logger.info("这是一条普通信息")
logger.warning("这是一条警告信息")
logger.error("这是一条错误信息")
logger.critical("这是一条严重错误信息")

运行后你会看到彩色的、带时间戳和文件位置的日志输出：

2024-03-28 10:30:00.123 | DEBUG    | __main__:<module>:3 - 这是一条调试信息
2024-03-28 10:30:00.124 | INFO     | __main__:<module>:4 - 这是一条普通信息
2024-03-28 10:30:00.125 | WARNING  | __main__:<module>:5 - 这是一条警告信息
2024-03-28 10:30:00.126 | ERROR    | __main__:<module>:6 - 这是一条错误信息
2024-03-28 10:30:00.127 | CRITICAL | __main__:<module>:7 - 这是一条严重错误信息

日志分级策略

在爬虫项目中，建议按以下方式使用日志级别：

级别	使用场景	爬虫示例
DEBUG	详细的调试信息，如请求参数、响应内容片段	请求URL、响应状态码、响应内容片段
INFO	正常的运行信息，如"开始爬取第N页"、"成功保存N条数据"	爬取进度、数据保存成功
WARNING	可恢复的异常，如"请求超时，准备重试"	请求超时、触发频率限制、需要重试
ERROR	严重错误但程序可继续，如"单个页面解析失败"	单个页面解析失败、数据格式异常
CRITICAL	致命错误，程序无法继续，如"数据库连接失败"	数据库连接失败、认证失效

日志持久化和轮转

在生产环境中，我们需要将日志保存到文件以便后续分析。loguru 提供了强大的日志文件管理功能：

from loguru import logger
import sys

# 移除默认的控制台输出（可选）
logger.remove()

# 添加控制台输出，只显示 INFO 及以上级别
logger.add(
    sys.stderr,
    level="INFO",
    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>"
)

# 添加文件输出，记录所有级别，按日期轮转
logger.add(
    "logs/crawler_{time:YYYY-MM-DD}.log",
    rotation="00:00",      # 每天午夜轮转
    retention="7 days",    # 保留7天的日志
    compression="zip",     # 旧日志压缩
    level="DEBUG",
    encoding="utf-8"
)

# 单独的错误日志文件
logger.add(
    "logs/error_{time:YYYY-MM-DD}.log",
    rotation="00:00",
    retention="30 days",
    level="ERROR",
    encoding="utf-8"
)

在爬虫中使用日志的最佳实践

通用示例

from loguru import logger

async def fetch_page(client, url: str) -> str:
    """获取页面内容"""
    logger.debug(f"准备请求: {url}")

    try:
        response = await client.get(url)
        response.raise_for_status()
        logger.info(f"请求成功: {url}, 状态码: {response.status_code}")
        return response.text
    except httpx.TimeoutException:
        logger.warning(f"请求超时: {url}")
        raise
    except httpx.HTTPStatusError as e:
        logger.error(f"HTTP错误: {url}, 状态码: {e.response.status_code}")
        raise
    except Exception as e:
        logger.exception(f"未知错误: {url}")  # exception() 会自动记录堆栈信息
        raise

爬虫日志实战示例

以下是一个名言网站爬虫的日志使用示例：

# quotes_crawler.py - 名言网站爬虫日志示例
from loguru import logger
import httpx

async def fetch_quotes(client: httpx.AsyncClient, page: int = 1):
    """
    获取名言列表

    展示在实际爬虫中如何使用日志
    """
    url = f"https://quotes.toscrape.com/page/{page}/"

    logger.debug(f"[QuotesCrawler] 请求参数: page={page}")

    try:
        response = await client.get(url)

        # HTTP状态码处理
        if response.status_code == 200:
            logger.info(f"[QuotesCrawler] 获取成功: 第{page}页")
            return response.text
        elif response.status_code == 404:
            logger.warning(f"[QuotesCrawler] 页面不存在: 第{page}页")
            return None
        elif response.status_code == 429:
            logger.warning(f"[QuotesCrawler] 触发频率限制，需要等待")
            raise Exception("触发频率限制")
        else:
            logger.error(f"[QuotesCrawler] HTTP错误: status_code={response.status_code}")
            return None

    except httpx.TimeoutException:
        logger.warning(f"[QuotesCrawler] 请求超时: {url}")
        raise
    except Exception as e:
        logger.exception(f"[QuotesCrawler] 未知错误: {e}")
        raise

日志前缀约定：在爬虫项目中，我们使用 [模块名] 作为日志前缀，便于在日志文件中快速定位问题来源，如 [QuotesCrawler]、[DataParser]、[DataStore]。

---

配置管理

为什么需要配置管理

硬编码的配置散落在代码各处是一种糟糕的实践：

# 反面示例 - 不要这样做
client = httpx.AsyncClient(timeout=30)
db_url = "mysql://root:password123@localhost/crawler"
API_KEY = "sk-xxxxxxxxxxxx"

这种做法的问题：

• 敏感信息（密码、API密钥）暴露在代码中
• 不同环境（开发、测试、生产）需要修改代码
• 配置分散，难以统一管理

pydantic-settings 简介

pydantic-settings 是 Pydantic 的扩展，专门用于处理应用配置。它支持：

• 从环境变量读取配置
• 从 .env 文件读取配置
• 配置值的类型验证
• 敏感信息的安全处理

安装：

pip install pydantic-settings

配置管理流程

基本使用

创建配置类：

# config/settings.py
from pydantic_settings import BaseSettings
from pydantic import Field
from typing import Optional


class CrawlerSettings(BaseSettings):
    """爬虫配置"""

    # 基础配置
    debug: bool = Field(default=False, description="调试模式")
    log_level: str = Field(default="INFO", description="日志级别")

    # 请求配置
    request_timeout: int = Field(default=30, description="请求超时时间(秒)")
    max_retries: int = Field(default=3, description="最大重试次数")
    retry_delay: float = Field(default=1.0, description="重试延迟(秒)")

    # 并发配置
    max_concurrent: int = Field(default=10, description="最大并发数")

    # 数据库配置
    db_host: str = Field(default="localhost", description="数据库主机")
    db_port: int = Field(default=3306, description="数据库端口")
    db_user: str = Field(default="root", description="数据库用户")
    db_password: str = Field(default="", description="数据库密码")
    db_name: str = Field(default="crawler", description="数据库名")

    # 代理配置
    proxy_url: Optional[str] = Field(default=None, description="代理地址")

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
        # 环境变量前缀，如 CRAWLER_DEBUG=true
        env_prefix = "CRAWLER_"


# 全局配置实例
settings = CrawlerSettings()

爬虫配置实战示例

以下是名言网站爬虫的配置示例，展示如何组织配置文件：

# config/settings.py - 爬虫配置
from pydantic_settings import BaseSettings
from pydantic import Field
from typing import Optional
from enum import Enum


class StorageType(str, Enum):
    """存储类型"""
    JSON = "json"
    CSV = "csv"


class CrawlerSettings(BaseSettings):
    """爬虫配置"""

    # 基础配置
    app_name: str = "QuotesCrawler"
    debug: bool = False
    log_level: str = "INFO"

    # 请求配置
    request_timeout: int = 30
    max_retries: int = 3
    retry_delay: float = 1.0

    # 爬虫配置
    base_url: str = "https://quotes.toscrape.com"
    max_pages: int = 10
    max_concurrency: int = 3
    crawl_delay_min: float = 0.5
    crawl_delay_max: float = 1.5

    # 存储配置
    storage_type: StorageType = StorageType.JSON
    output_dir: str = "./output"

    class Config:
        env_file = ".env"
        env_prefix = "CRAWLER_"


# 全局配置实例
settings = CrawlerSettings()

# config/constants.py - 常量配置
"""爬虫常量配置"""

# 默认请求头
DEFAULT_HEADERS = {
    "User-Agent": "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",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}

# HTTP 状态码
HTTP_OK = 200
HTTP_NOT_FOUND = 404
HTTP_TOO_MANY_REQUESTS = 429
HTTP_SERVER_ERROR = 500

使用 .env 文件

创建 .env 文件（记得添加到 .gitignore）：

# .env
CRAWLER_DEBUG=true
CRAWLER_LOG_LEVEL=DEBUG
CRAWLER_REQUEST_TIMEOUT=60
CRAWLER_DB_PASSWORD=your_secret_password
CRAWLER_PROXY_URL=http://127.0.0.1:7890

在代码中使用配置：

from config.settings import settings
from loguru import logger

# 根据配置设置日志级别
logger.add("logs/app.log", level=settings.log_level)

# 使用配置创建客户端
client = httpx.AsyncClient(
    timeout=settings.request_timeout,
    proxies=settings.proxy_url
)

logger.info(f"爬虫启动，调试模式: {settings.debug}")

多环境配置

对于需要区分开发、测试、生产环境的项目，可以这样组织：

# config/settings.py
import os
from pydantic_settings import BaseSettings
from pydantic import Field


class BaseConfig(BaseSettings):
    """基础配置"""
    env: str = Field(default="development", description="运行环境")
    debug: bool = False
    log_level: str = "INFO"

    class Config:
        env_file = ".env"
        env_prefix = "CRAWLER_"


class DevelopmentConfig(BaseConfig):
    """开发环境配置"""
    debug: bool = True
    log_level: str = "DEBUG"


class ProductionConfig(BaseConfig):
    """生产环境配置"""
    debug: bool = False
    log_level: str = "WARNING"


def get_settings() -> BaseConfig:
    """根据环境变量返回对应的配置"""
    env = os.getenv("CRAWLER_ENV", "development")
    config_map = {
        "development": DevelopmentConfig,
        "production": ProductionConfig,
    }
    config_class = config_map.get(env, DevelopmentConfig)
    return config_class()


settings = get_settings()

---

异常处理与重试机制

异常处理流程

自定义异常类

为爬虫项目定义专门的异常类，便于区分和处理不同类型的错误：

# exceptions.py

class CrawlerException(Exception):
    """爬虫基础异常类"""
    pass


class RequestException(CrawlerException):
    """请求相关异常"""
    pass


class ParseException(CrawlerException):
    """解析相关异常"""
    pass


class StorageException(CrawlerException):
    """存储相关异常"""
    pass


class RateLimitException(RequestException):
    """触发速率限制"""
    pass


class IPBlockedException(RequestException):
    """IP被封禁"""
    pass


class LoginRequiredException(CrawlerException):
    """需要登录"""
    pass

爬虫异常定义实战

以下是通用爬虫的异常类设计示例：

# exceptions/crawler_exceptions.py
"""爬虫异常类"""


class CrawlerException(Exception):
    """爬虫基础异常"""
    pass


class RequestException(CrawlerException):
    """请求相关异常"""
    pass


class ParseException(CrawlerException):
    """解析相关异常"""
    pass


class StorageException(CrawlerException):
    """存储相关异常"""
    pass


class RateLimitException(RequestException):
    """触发速率限制"""
    pass


class IPBlockedException(RequestException):
    """IP被封禁"""
    pass


class PageNotFoundException(RequestException):
    """页面不存在"""
    pass


# HTTP 状态码映射
HTTP_ERROR_MESSAGES = {
    400: "请求参数错误",
    401: "未授权访问",
    403: "禁止访问",
    404: "页面不存在",
    429: "请求过于频繁",
    500: "服务器内部错误",
    502: "网关错误",
    503: "服务暂时不可用",
}

使用 tenacity 实现重试

tenacity 是一个强大的重试库，比手写重试逻辑更加优雅和可靠。

安装：

pip install tenacity

基本使用：

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log
)
from loguru import logger
import httpx

from exceptions import RequestException, RateLimitException


@retry(
    stop=stop_after_attempt(3),                    # 最多重试3次
    wait=wait_exponential(multiplier=1, max=10),   # 指数退避，最长等待10秒
    retry=retry_if_exception_type((                # 只对特定异常重试
        httpx.TimeoutException,
        httpx.ConnectError,
        RequestException
    )),
    before_sleep=before_sleep_log(logger, "WARNING")  # 重试前记录日志
)
async def fetch_with_retry(client: httpx.AsyncClient, url: str) -> str:
    """带重试的请求函数"""
    response = await client.get(url)

    # 检查是否触发速率限制
    if response.status_code == 429:
        raise RateLimitException("触发速率限制")

    response.raise_for_status()
    return response.text

更复杂的重试策略

from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
    retry_if_exception_type,
    RetryError
)


def create_retry_decorator(max_attempts: int = 3, max_wait: int = 60):
    """创建可配置的重试装饰器"""
    return retry(
        stop=stop_after_attempt(max_attempts),
        wait=wait_random_exponential(multiplier=1, max=max_wait),
        retry=retry_if_exception_type((
            httpx.TimeoutException,
            httpx.ConnectError,
            httpx.HTTPStatusError,
        )),
        reraise=True  # 重试用尽后重新抛出异常
    )


# 使用
@create_retry_decorator(max_attempts=5, max_wait=30)
async def fetch_important_data(client, url):
    """重要数据获取，使用更多重试次数"""
    response = await client.get(url)
    response.raise_for_status()
    return response.json()

全局异常处理

在爬虫主程序中实现全局异常处理：

import asyncio
from loguru import logger
from exceptions import CrawlerException, IPBlockedException


async def run_crawler():
    """爬虫主程序"""
    try:
        # 爬虫逻辑
        await crawl_all_pages()
    except IPBlockedException as e:
        logger.critical(f"IP被封禁，程序终止: {e}")
        # 可以在这里触发告警通知
        raise
    except CrawlerException as e:
        logger.error(f"爬虫异常: {e}")
        raise
    except asyncio.CancelledError:
        logger.warning("任务被取消")
        raise
    except Exception as e:
        logger.exception(f"未预期的异常: {e}")
        raise
    finally:
        logger.info("爬虫任务结束，执行清理工作...")
        # 清理资源
        await cleanup()


if __name__ == "__main__":
    try:
        asyncio.run(run_crawler())
    except KeyboardInterrupt:
        logger.info("用户中断程序")
    except Exception as e:
        logger.critical(f"程序异常退出: {e}")
        exit(1)

---

项目目录结构

项目架构图

推荐的目录结构

一个工程化的爬虫项目应该有清晰的目录结构：

my_crawler/
├── config/                 # 配置模块
│   ├── __init__.py
│   └── settings.py         # 配置定义
├── core/                   # 核心模块
│   ├── __init__.py
│   ├── client.py          # HTTP客户端封装
│   └── retry.py           # 重试策略
├── crawler/               # 爬虫模块
│   ├── __init__.py
│   ├── base.py            # 爬虫基类
│   └── xxx_crawler.py     # 具体爬虫实现
├── parser/                # 解析模块
│   ├── __init__.py
│   └── xxx_parser.py      # 页面解析器
├── store/                 # 存储模块
│   ├── __init__.py
│   ├── base.py            # 存储基类
│   ├── mysql.py           # MySQL存储
│   └── json_store.py      # JSON文件存储
├── models/                # 数据模型
│   ├── __init__.py
│   └── xxx_model.py       # Pydantic模型定义
├── exceptions/            # 异常定义
│   ├── __init__.py
│   └── crawler_exceptions.py
├── utils/                 # 工具函数
│   ├── __init__.py
│   └── helpers.py
├── logs/                  # 日志目录
├── data/                  # 数据输出目录
├── tests/                 # 测试目录
│   └── test_xxx.py
├── .env                   # 环境变量（不提交到git）
├── .env.example           # 环境变量示例
├── .gitignore
├── requirements.txt       # 依赖列表
├── main.py               # 程序入口
└── README.md             # 项目说明

模块划分原则

1. 单一职责：每个模块只负责一件事
2. 高内聚低耦合：相关代码放在一起，模块间依赖最小化
3. 抽象与实现分离：定义基类/接口，方便扩展和替换

---

实战案例：工程化改造

让我们将入门教程中的爬虫进行工程化改造。

改造前（入门教程版本）

# 入门教程的简单版本
import httpx
from parsel import Selector

async def crawl():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://example.com")
        print(f"状态码: {response.status_code}")
        # 解析...
        # 存储...

改造后（工程化版本）

项目结构：

refactored_crawler/
├── config/
│   ├── __init__.py
│   └── settings.py
├── exceptions.py
├── logger_config.py
├── client.py
├── crawler.py
└── main.py

核心代码示例：

# logger_config.py
import sys
from loguru import logger
from config.settings import settings


def setup_logger():
    """配置日志系统"""
    # 移除默认处理器
    logger.remove()

    # 控制台输出
    logger.add(
        sys.stderr,
        level=settings.log_level,
        format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
               "<level>{level: <8}</level> | "
               "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
               "<level>{message}</level>"
    )

    # 文件输出
    logger.add(
        f"logs/crawler_{settings.env}.log",
        rotation="00:00",
        retention="7 days",
        level="DEBUG",
        encoding="utf-8"
    )

    return logger

# client.py
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
from loguru import logger
from config.settings import settings
from exceptions import RequestException


class CrawlerClient:
    """封装的HTTP客户端"""

    def __init__(self):
        self.client = httpx.AsyncClient(
            timeout=settings.request_timeout,
            headers={
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
            }
        )

    @retry(
        stop=stop_after_attempt(settings.max_retries),
        wait=wait_exponential(multiplier=1, max=10)
    )
    async def get(self, url: str) -> httpx.Response:
        """发送GET请求，带重试"""
        logger.debug(f"请求: {url}")
        response = await self.client.get(url)
        response.raise_for_status()
        logger.info(f"请求成功: {url}")
        return response

    async def close(self):
        """关闭客户端"""
        await self.client.aclose()
        logger.debug("HTTP客户端已关闭")

# main.py
import asyncio
from loguru import logger
from logger_config import setup_logger
from crawler import BBSCrawler
from config.settings import settings


async def main():
    """主程序入口"""
    # 初始化日志
    setup_logger()

    logger.info(f"爬虫启动 - 环境: {settings.env}, 调试模式: {settings.debug}")

    crawler = BBSCrawler()
    try:
        await crawler.run()
    except Exception as e:
        logger.exception(f"爬虫运行异常: {e}")
        raise
    finally:
        await crawler.cleanup()
        logger.info("爬虫运行结束")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("用户中断")

---

本章小结

本章我们学习了爬虫工程化开发的核心内容：

1. 日志系统：使用 loguru 替代 print，实现分级日志、日志轮转和持久化
2. 配置管理：使用 pydantic-settings 统一管理配置，支持环境变量和 .env 文件
3. 异常处理：自定义异常类，使用 tenacity 实现优雅的重试机制
4. 项目结构：遵循单一职责原则，合理划分模块

这些工程化实践将贯穿整个进阶教程，是后续学习的基础。

---

下一章预告

下一章我们将学习「反爬虫对抗基础：请求伪装」。主要内容包括：

• 常见的反爬虫检测手段
• User-Agent 随机轮换
• 完整的请求头伪装
• 使用 curl_cffi 模拟浏览器指纹
• 智能速率控制

这将是我们与反爬虫斗争的第一步，让我们的爬虫不再那么容易被识别和封禁。