㊗️本期内容已收录至专栏《Python爬虫实战》，持续完善知识体系与项目实战，建议先订阅收藏，后续查阅更方便～
㊙️本期爬虫难度指数：⭐⭐☆☆☆（基础级）
🉐福利： 一次订阅后，专栏内的所有文章可永久免费看，持续更新中，保底1000+(篇)硬核实战内容。

全文目录：

🌟 开篇语

哈喽，各位小伙伴们你们好呀～我是【喵手】。
运营社区： C站 / 掘金 / 腾讯云 / 阿里云 / 华为云 / 51CTO
欢迎大家常来逛逛，一起学习，一起进步～🌟

  我长期专注 Python 爬虫工程化实战，主理专栏👉 《Python爬虫实战》：从采集策略到反爬对抗，从数据清洗到分布式调度，持续输出可复用的方法论与可落地案例。内容主打一个“能跑、能用、能扩展”，让数据价值真正做到——抓得到、洗得净、用得上。

  📌 专栏食用指南（建议收藏）

  • ✅ 入门基础：环境搭建 / 请求与解析 / 数据落库
  • ✅ 进阶提升：登录鉴权 / 动态渲染 / 反爬对抗
  • ✅ 工程实战：异步并发 / 分布式调度 / 监控与容错
  • ✅ 项目落地：数据治理 / 可视化分析 / 场景化应用

📣 专栏推广时间：如果你想系统学爬虫，而不是碎片化东拼西凑，欢迎订阅专栏👉《Python爬虫实战》👈，一次订阅后，专栏内的所有文章可永久免费阅读，持续更新中。
  
💕订阅后更新会优先推送，按目录学习更高效💯～

0️⃣ 前言（Preface）

这篇文章要做一件很具体的事：用 Python 抓取一个公开的农作物病虫害图谱类网站，从分类/搜索页拿到详情页链接，再进入详情页解析“病虫害名、作物、症状摘要、防治说明、图片链接”，最后导出为 CSV 与 JSON 文件。

读完这篇文章，你至少能获得三样东西：

第一，你会掌握“列表页 + 详情页”的二段式采集思路，而不是只停留在单页解析。

第二，你会得到一套可运行的 Python 项目结构，包含请求层、解析层、存储层、日志、重试、去重和命令行入口。

第三，你会知道在农作物病虫害这类知识型网页上，如何处理字段不齐、图片懒加载、页面轻度动态渲染、403/429、编码乱码等实际问题。

我个人比较喜欢这类案例。它不像电商价格那样变化剧烈，也不像登录站点那样容易踩边界；它更像一个信息整理工程：把散落在网页中的农业知识结构化，后续可以用于检索、数据分析、知识库建设，甚至可以作为图像识别数据标注前的目录基础。当然，这里只讨论公开网页的低频合规采集，不讨论绕过登录、付费墙、验证码或任何不合适的访问方式。

1️⃣ 摘要（Abstract）

本文以农作物病虫害图谱目录页为案例，使用 Python、requests、BeautifulSoup、lxml、tenacity、pandas 等工具，实现“分类页采集详情链接 → 详情页抽取字段 → 清洗去重 → 导出 CSV/JSON”的完整流程，最终产出一份结构化病虫害数据表。

读完后，你可以学到：

  1. 如何设计一个可维护的二段式爬虫，而不是把所有逻辑塞进一个脚本里。
  2. 如何从列表页提取详情链接，再到详情页解析病虫害名称、作物、症状摘要、防治说明和图片链接。
  3. 如何处理请求失败、字段缺失、URL 去重、图片地址规范化、输出编码和日志排错。

本文会尽量把代码写成真实项目的样子。你可以直接复制到本地运行，也可以把目标站点替换成其他农业知识站点，只需要调整解析器里的 CSS 选择器或 XPath 即可。

2️⃣ 背景与需求（Why）

2.1 为什么要爬农作物病虫害图谱

农作物病虫害信息通常分散在农业技术网站、科研机构网站、植保站页面、病虫害知识库、PDF 技术手册和图谱系统中。人工查询当然可以，但当我们要做批量整理时，人工复制粘贴会非常低效，而且容易漏字段、漏图片、漏来源链接。

把公开病虫害图谱页面结构化之后，可以做很多事情：

一是做信息聚合。比如把水稻、玉米、番茄、柑橘等作物相关病虫害汇总成统一表格，方便农业技术人员、内容编辑或数据分析人员查阅。

二是做数据分析。比如统计某种作物常见病害数量、不同病虫害涉及的作物分布、关键词高频症状、防治措施文本长度等。

三是做知识库建设。结构化字段可以进一步进入 SQLite、MySQL、Elasticsearch 或向量数据库，用于检索问答、后台管理系统或知识图谱原型。

四是做图片数据索引。图谱类网站常包含病斑、虫体、受害叶片、果实受害等图片。我们不一定要立即下载图片，但至少可以先保存图片链接、详情页来源和文字说明，为后续合法使用、人工审核和数据标注做准备。

需要特别说明：病虫害防治涉及农业生产安全，本文抓取的数据仅用于技术学习与信息整理，不应直接替代当地农技人员、植保专家或官方登记用药说明。实际生产中，防治措施要结合当地法规、作物生育期、农药登记情况、安全间隔期和田间实际诊断。

2.2 目标站点与目标字段

本文以公开的农作物病虫害图谱/查询类网页为示例。实际项目中，我建议先从下面这些页面形态入手：

  • 病虫害列表页
  • 病虫害搜索页
  • 按作物分类的目录页
  • 按病害、虫害、害螨、益虫分类的目录页
  • 病虫害详情页
  • 防治技术详情页

本文的目标字段如下：

字段名 含义 说明
pest_name 病虫害名 如稻热病、白叶枯病、二化螟、蚜虫等
crop 作物 如水稻、玉米、番茄、柑橘、茶等
symptom_summary 症状摘要 病斑、叶片受害、茎秆受害、虫害危害表现等
control_method 防治说明 农业防治、物理防治、生物防治、化学防治、综合防治等
image_url 图片链接 详情页中的主图或图集图片链接
detail_url 详情页链接 用于溯源和去重
source_site 来源站点 方便后续多站点合并
crawled_at 抓取时间 数据快照时间

用户要求的核心字段是“病虫害名、作物、症状摘要、防治说明、图片链接”。我额外加了 detail_url、source_site、crawled_at，是因为真实爬虫项目里必须考虑溯源、去重和数据版本管理。没有来源 URL 的爬虫数据，后期排查会非常痛苦。

3️⃣ 合规与注意事项

写爬虫之前，先把边界说清楚。技术本身没有问题，关键在于怎么用。

3.1 robots.txt 基本说明

robots.txt 是网站放在根目录下的爬虫访问声明文件，通常用于告诉搜索引擎和自动化程序哪些路径可以访问，哪些路径不希望访问。

一个典型的 robots.txt 可能长这样：

User-agent: *
Disallow: /admin/
Disallow: /login/
Allow: /
Crawl-delay: 5

含义大致是：

  • User-agent: * 表示对所有爬虫生效。
  • Disallow 表示不希望爬虫访问的路径。
  • Allow 表示允许访问的路径。
  • Crawl-delay 表示建议访问间隔。

在实际采集前，建议先访问：

https://目标域名/robots.txt

如果 robots.txt 明确禁止某些路径，就不要采集这些路径。如果目标站没有 robots.txt，也不等于可以高频抓取，只能说明没有通过这个文件表达规则，依然要保持低频、少量、合规和可追踪。

在本文代码中，我会把 robots 检查做成一个提示函数：它不会强行替你判断所有法律与许可问题，但会提醒你采集前先检查站点规则。

3.2 控制频率，不要攻击式并发

爬虫最容易出问题的地方不是解析，而是请求频率。很多人一上来就开几十个线程，短时间内访问几千次页面，这种做法既不专业，也容易给对方服务器造成负担。

本文建议：

  • 默认每次请求之间 sleep 1 到 3 秒。
  • 单机学习项目先不要上高并发。
  • 请求失败时做指数退避，而不是立即疯狂重试。
  • 只抓自己需要的字段，不做全站无差别扫描。
  • 测试阶段先限制页数，例如只抓前 20 条详情。
  • 每次抓取都保留日志，方便出现问题时停止或回滚。

我自己的习惯是：先用 5 到 10 条数据验证解析逻辑，再扩大到几十条，最后才考虑更多页面。爬虫不是越快越好，稳定、可控、可解释才是更好的工程标准。

3.3 不采集敏感信息，不绕过限制

本文只面向公开页面的技术学习，不涉及以下行为：

  • 不采集个人隐私数据。
  • 不采集账号、手机号、邮箱等敏感信息。
  • 不绕过登录限制。
  • 不绕过付费墙。
  • 不破解验证码。
  • 不伪装成真实用户进行高频访问。
  • 不访问后台、管理端、接口密钥或非公开资源。
  • 不对目标站点做压力测试。

如果目标页面需要登录才能看到，或者数据明显属于非公开范围，就不应继续抓取。技术上能不能做到，不等于应该做。

4️⃣ 技术选型与整体流程（What / How）

4.1 静态、动态、API 三种页面

写爬虫之前，先判断页面属于哪类。

第一类是静态页面。浏览器看到的内容，在 requests 获取的 HTML 里也能找到。这种页面最好处理，用 requests + BeautifulSoup / lxml 就够了。

第二类是动态页面。requests 获取到的 HTML 是空壳，真正内容由浏览器执行 JavaScript 后渲染出来。这类页面可以考虑 Playwright、Selenium，也可以在开发者工具里寻找接口。

第三类是 API 页面。网页内容其实来自 JSON 接口，只要找到接口参数，直接请求 API 就能拿到结构化数据。这类最干净，但也要注意接口是否公开、是否有频控、是否允许自动访问。

本文的案例属于“列表页 + 详情页”的二段式抓取。页面可能存在轻度动态渲染，因此我们采用一个比较稳妥的方案：

  • 默认使用 requests 获取页面。
  • 使用 BeautifulSoup + lxml 解析 HTML。
  • 如果列表页拿不到详情链接，可以启用 Playwright 渲染兜底。
  • 存储使用 CSV 和 JSON Lines，便于查看和后续导入数据库。
  • 项目保留 SQLite 扩展位置，但起步不强依赖数据库。

4.2 整体流程

整体流程可以写成下面这样：

读取配置
   ↓
检查 robots.txt 与目标域名
   ↓
构造列表页 / 搜索页 URL
   ↓
请求列表页 HTML
   ↓
解析详情页链接
   ↓
详情页 URL 去重
   ↓
逐个请求详情页
   ↓
解析字段：病虫害名、作物、症状、防治、图片
   ↓
字段清洗与容错
   ↓
按 detail_url 去重
   ↓
导出 CSV / JSONL
   ↓
输出日志与统计信息

对应到代码模块：

crop_pest_spider/
├── config.py          # 配置
├── fetcher.py         # 请求层
├── parser.py          # 解析层
├── storage.py         # 存储层
├── models.py          # 数据模型
├── utils.py           # 工具函数
├── main.py            # 入口文件
├── requirements.txt   # 依赖
└── data/
    ├── raw/           # 原始 HTML，可选
    └── output/        # 输出 CSV/JSONL

4.3 为什么选择 requests + BeautifulSoup + lxml

这个项目不一上来就用 Scrapy，原因很简单：本文重点是讲清楚二段式抓取，不希望框架本身分散注意力。

requests 适合处理请求、Session、Header、Timeout。

BeautifulSoup 适合容错解析，网页结构不规整时也比较舒服。

lxml 解析速度快，配合 BeautifulSoup 使用时可以兼顾性能和易读性。

tenacity 用于重试和退避，比自己手写 while 循环更干净。

pandas 用于导出 CSV，编码、列顺序和缺失值处理更方便。

如果后续要扩大规模，再迁移到 Scrapy；如果确认页面强依赖 JavaScript，再把列表页或详情页切到 Playwright。工程上没有必要一开始就把工具堆满，先用小而清楚的方案跑通闭环，通常更稳。

5️⃣ 环境准备与依赖安装

5.1 Python 版本

建议使用 Python 3.10 或以上版本。本文代码在 Python 3.10、3.11、3.12 下都可以运行。

查看版本：

python --version

如果你的系统里同时有多个 Python，可以使用：

python3 --version

5.2 创建虚拟环境

Windows：

mkdir crop_pest_spider
cd crop_pest_spider

python -m venv .venv
.venv\Scripts\activate

macOS / Linux：

mkdir crop_pest_spider
cd crop_pest_spider

python3 -m venv .venv
source .venv/bin/activate

5.3 安装依赖

新建 requirements.txt：

requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.2.2
pandas==2.2.2
tenacity==8.5.0
python-dateutil==2.9.0.post0
tqdm==4.66.4

安装：

pip install -r requirements.txt

如果你要启用 Playwright 兜底，可以额外安装：

pip install playwright==1.45.0
python -m playwright install chromium

本文主代码不强制依赖 Playwright。只有当 requests 抓不到渲染后的列表时，才建议补上。

5.4 推荐项目结构

创建目录：

mkdir -p crop_pest_spider/data/raw
mkdir -p crop_pest_spider/data/output
touch crop_pest_spider/__init__.py
touch crop_pest_spider/config.py
touch crop_pest_spider/models.py
touch crop_pest_spider/utils.py
touch crop_pest_spider/fetcher.py
touch crop_pest_spider/parser.py
touch crop_pest_spider/storage.py
touch crop_pest_spider/main.py

Windows PowerShell 可以这样创建：

mkdir crop_pest_spider
mkdir crop_pest_spider\data
mkdir crop_pest_spider\data\raw
mkdir crop_pest_spider\data\output
New-Item crop_pest_spider\__init__.py
New-Item crop_pest_spider\config.py
New-Item crop_pest_spider\models.py
New-Item crop_pest_spider\utils.py
New-Item crop_pest_spider\fetcher.py
New-Item crop_pest_spider\parser.py
New-Item crop_pest_spider\storage.py
New-Item crop_pest_spider\main.py

6️⃣ 核心实现：请求层（Fetcher）

请求层要解决的问题不是“能不能请求成功”这么简单，而是要让请求行为稳定、克制、可追踪。

本文请求层包含以下能力：

  • 设置 User-Agent。
  • 设置 Referer。
  • 设置 timeout。
  • 使用 requests.Session 复用连接。
  • 支持 Cookie，但默认不使用。
  • 对 429、500、502、503、504 做重试。
  • 使用指数退避。
  • 每次请求之间随机 sleep。
  • 保存少量原始 HTML，方便排错。

6.1 配置文件 config.py

# crop_pest_spider/config.py
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
RAW_DIR = DATA_DIR / "raw"
OUTPUT_DIR = DATA_DIR / "output"

RAW_DIR.mkdir(parents=True, exist_ok=True)
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

SOURCE_SITE = "农业病虫害智能管理决策系统"

# 示例目标站点。实际使用时可替换为你确认允许采集的公开站点。
BASE_URL = "https://azai.tari.gov.tw"

# 搜索页模板：term 可以换成 水稻、玉米、番茄、柑桔 等关键词。
SEARCH_URL_TEMPLATE = BASE_URL + "/search.html?stype=0&term={keyword}"

# 默认关键词。建议先小批量测试，不要一上来全量扫描。
DEFAULT_KEYWORDS = [
    "水稻",
    "玉米",
    "番茄",
]

# 请求头。UA 不要写得太离谱，保持普通浏览器风格即可。
DEFAULT_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0 Safari/537.36"
    ),
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,zh-TW;q=0.8,en;q=0.7",
    "Connection": "keep-alive",
}

REQUEST_TIMEOUT = 15

# 访问间隔。学习项目建议保守一点。
MIN_DELAY = 1.2
MAX_DELAY = 3.5

# 重试配置。
RETRY_TIMES = 3
RETRY_STATUS_CODES = {429, 500, 502, 503, 504}

# 测试阶段最多抓多少条详情。None 表示不限制。
DEFAULT_MAX_DETAILS = 30

CSV_FILE = OUTPUT_DIR / "crop_pest_records.csv"
JSONL_FILE = OUTPUT_DIR / "crop_pest_records.jsonl"

# 是否保存原始 HTML。调试选择器时很有用，正式大量运行时可关闭。
SAVE_RAW_HTML = True

6.2 数据模型 models.py

用 dataclass 管理字段，比直接传字典更清楚。

# crop_pest_spider/models.py
from dataclasses import dataclass, asdict
from typing import Optional


@dataclass
class PestRecord:
    pest_name: str = ""
    crop: str = ""
    symptom_summary: str = ""
    control_method: str = ""
    image_url: str = ""
    detail_url: str = ""
    source_site: str = ""
    crawled_at: str = ""

    def to_dict(self) -> dict:
        return asdict(self)

    def is_valid(self) -> bool:
        """
        最基本的数据有效性判断。
        真实项目可以更严格，比如 pest_name 和 detail_url 必须存在。
        """
        return bool(self.pest_name or self.detail_url)

6.3 工具函数 utils.py

# crop_pest_spider/utils.py
import hashlib
import re
import time
import random
import logging
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urljoin, urlparse


def setup_logger(name: str = "crop_pest_spider") -> logging.Logger:
    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)

    if logger.handlers:
        return logger

    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

    return logger


logger = setup_logger()


def random_sleep(min_delay: float, max_delay: float) -> None:
    delay = random.uniform(min_delay, max_delay)
    time.sleep(delay)


def normalize_space(text: str) -> str:
    if not text:
        return ""
    text = text.replace("\xa0", " ")
    text = re.sub(r"\s+", " ", text)
    return text.strip()


def normalize_url(base_url: str, href: str) -> str:
    if not href:
        return ""
    return urljoin(base_url, href.strip())


def now_iso() -> str:
    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


def safe_filename_from_url(url: str, suffix: str = ".html") -> str:
    parsed = urlparse(url)
    raw = f"{parsed.netloc}_{parsed.path}_{parsed.query}"
    name = hashlib.md5(raw.encode("utf-8")).hexdigest()
    return f"{name}{suffix}"


def content_hash(*values: str) -> str:
    raw = "||".join(v or "" for v in values)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def save_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")

6.4 请求层 fetcher.py

# crop_pest_spider/fetcher.py
from typing import Optional
import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

from .config import (
    DEFAULT_HEADERS,
    REQUEST_TIMEOUT,
    RETRY_TIMES,
    RETRY_STATUS_CODES,
    MIN_DELAY,
    MAX_DELAY,
    RAW_DIR,
    SAVE_RAW_HTML,
)
from .utils import logger, random_sleep, safe_filename_from_url, save_text


class FetchError(Exception):
    pass


class Fetcher:
    def __init__(
        self,
        headers: Optional[dict] = None,
        timeout: int = REQUEST_TIMEOUT,
        min_delay: float = MIN_DELAY,
        max_delay: float = MAX_DELAY,
        save_raw_html: bool = SAVE_RAW_HTML,
    ):
        self.session = requests.Session()
        self.headers = headers or DEFAULT_HEADERS.copy()
        self.timeout = timeout
        self.min_delay = min_delay
        self.max_delay = max_delay
        self.save_raw_html = save_raw_html

        self.session.headers.update(self.headers)

    def set_cookie(self, cookie: str) -> None:
        """
        如目标公开页面不需要 cookie，就不要设置。
        如果你在合规场景下需要使用自己的会话 cookie，可以通过这里设置。
        """
        if cookie:
            self.session.headers.update({"Cookie": cookie})

    def _should_retry_by_status(self, status_code: int) -> bool:
        return status_code in RETRY_STATUS_CODES

    @retry(
        reraise=True,
        stop=stop_after_attempt(RETRY_TIMES),
        wait=wait_exponential(multiplier=1, min=2, max=12),
        retry=retry_if_exception_type(FetchError),
    )
    def get(self, url: str, referer: Optional[str] = None) -> str:
        headers = self.headers.copy()
        if referer:
            headers["Referer"] = referer

        logger.info("GET %s", url)

        try:
            resp = self.session.get(
                url,
                headers=headers,
                timeout=self.timeout,
            )
        except requests.RequestException as exc:
            raise FetchError(f"request failed: {url}, error={exc}") from exc

        if self._should_retry_by_status(resp.status_code):
            raise FetchError(f"retryable status={resp.status_code}, url={url}")

        if resp.status_code == 403:
            raise FetchError(
                f"403 Forbidden: {url}. "
                f"建议降低频率、检查 robots.txt、确认页面是否允许公开访问。"
            )

        if resp.status_code == 404:
            logger.warning("404 Not Found: %s", url)
            return ""

        if not resp.ok:
            raise FetchError(f"unexpected status={resp.status_code}, url={url}")

        # requests 通常会根据响应头判断编码；如果网页中文乱码，可手动兜底。
        if not resp.encoding or resp.encoding.lower() == "iso-8859-1":
            resp.encoding = resp.apparent_encoding

        html = resp.text

        if self.save_raw_html:
            filename = safe_filename_from_url(url)
            save_text(RAW_DIR / filename, html)

        random_sleep(self.min_delay, self.max_delay)
        return html

    def close(self) -> None:
        self.session.close()

这里有几个小细节值得解释。

首先，timeout 一定要写。没有 timeout 的 requests 在网络抖动时可能卡很久。

其次，失败重试只针对可重试状态，比如 429、500、502、503、504。403 不建议硬重试，因为它通常代表访问被拒绝，再重试也没有意义。

第三，Referer 在详情页请求时可以设置成列表页 URL，模拟正常页面跳转路径。但这不代表可以绕过限制，只是让请求更接近普通浏览行为。

第四，保存原始 HTML 对排错很重要。解析器失效时，先看 raw HTML，而不是凭空猜网页结构。

7️⃣ 核心实现：解析层（Parser）

解析层是这类项目最容易写乱的地方。我的建议是：列表页解析和详情页解析分开写，每个函数只做一件事。

7.1 解析策略

本文采用 BeautifulSoup + CSS 选择器。

列表页主要任务：

  • 找到所有详情页链接。
  • 过滤无关链接。
  • 补全相对 URL。
  • 去重。
  • 可选提取列表页上的简短摘要。

详情页主要任务：

  • 提取病虫害名。
  • 提取作物。
  • 提取症状摘要。
  • 提取防治说明。
  • 提取图片链接。
  • 字段缺失时返回空字符串，而不是直接报错。

7.2 解析层 parser.py

# crop_pest_spider/parser.py
import re
from typing import List, Iterable
from urllib.parse import urlparse, parse_qs, unquote

from bs4 import BeautifulSoup

from .config import BASE_URL, SOURCE_SITE
from .models import PestRecord
from .utils import normalize_space, normalize_url, now_iso, logger


DETAIL_URL_PATTERNS = [
    "/search/bug/detail",
    "/search/bug/detail2",
]


def make_soup(html: str) -> BeautifulSoup:
    return BeautifulSoup(html or "", "lxml")


def is_detail_url(url: str) -> bool:
    if not url:
        return False
    return any(pattern in url for pattern in DETAIL_URL_PATTERNS)


def unique_keep_order(items: Iterable[str]) -> List[str]:
    seen = set()
    result = []
    for item in items:
        if not item:
            continue
        if item in seen:
            continue
        seen.add(item)
        result.append(item)
    return result


def parse_list_page(html: str, base_url: str = BASE_URL) -> List[str]:
    """
    从列表页/搜索页提取详情页链接。
    这个函数故意写得宽松一点，因为不同图谱站点的 class 名经常会变。
    """
    soup = make_soup(html)
    links = []

    for a in soup.select("a[href]"):
        href = a.get("href", "").strip()
        full_url = normalize_url(base_url, href)

        if is_detail_url(full_url):
            links.append(full_url)

    links = unique_keep_order(links)
    logger.info("parsed %d detail links from list page", len(links))
    return links


def text_by_selectors(soup: BeautifulSoup, selectors: List[str]) -> str:
    """
    按多个选择器尝试取文本，谁先命中就用谁。
    """
    for selector in selectors:
        node = soup.select_one(selector)
        if node:
            text = normalize_space(node.get_text(" ", strip=True))
            if text:
                return text
    return ""


def texts_by_selectors(soup: BeautifulSoup, selectors: List[str]) -> List[str]:
    result = []
    for selector in selectors:
        for node in soup.select(selector):
            text = normalize_space(node.get_text(" ", strip=True))
            if text:
                result.append(text)
    return unique_keep_order(result)


def extract_by_label(soup: BeautifulSoup, labels: List[str], max_chars: int = 3000) -> str:
    """
    尝试从包含固定标签的文本块中抽取字段。
    例如页面里可能出现：
    症状：叶片产生褐色病斑……
    防治方法：选用抗病品种……
    """
    full_text = normalize_space(soup.get_text("\n", strip=True))
    if not full_text:
        return ""

    # 用换行更方便做标签定位
    lines = [normalize_space(x) for x in full_text.split("\n") if normalize_space(x)]

    for i, line in enumerate(lines):
        for label in labels:
            if line.startswith(label):
                value = line.replace(label, "", 1)
                value = value.lstrip("：: ")
                if value:
                    return value[:max_chars]

                # 如果标签单独成行，取后面几行作为内容
                tail = " ".join(lines[i + 1 : i + 5])
                return tail[:max_chars]

    # 再做一次宽松正则
    label_pattern = "|".join(re.escape(label.rstrip("：:")) for label in labels)
    pattern = rf"({label_pattern})[：:\s]+(.{{1,{max_chars}}})"
    match = re.search(pattern, full_text)
    if match:
        return normalize_space(match.group(2))[:max_chars]

    return ""


def extract_name_from_url(detail_url: str) -> str:
    """
    detail2?name=稻熱病&plant=水稻 这类 URL 可以直接从 query 中兜底提取名称。
    """
    parsed = urlparse(detail_url)
    query = parse_qs(parsed.query)
    name_values = query.get("name") or []
    if name_values:
        return normalize_space(unquote(name_values[0]))
    return ""


def extract_crop_from_url(detail_url: str) -> str:
    parsed = urlparse(detail_url)
    query = parse_qs(parsed.query)
    plant_values = query.get("plant") or []
    if plant_values:
        return normalize_space(unquote(plant_values[0]))
    return ""


def extract_images(soup: BeautifulSoup, detail_url: str, base_url: str = BASE_URL) -> List[str]:
    """
    提取详情页图片。
    兼容普通 src、懒加载 data-src、data-original。
    """
    image_urls = []

    for img in soup.select("img"):
        src = (
            img.get("src")
            or img.get("data-src")
            or img.get("data-original")
            or img.get("data-lazy-src")
            or ""
        ).strip()

        if not src:
            continue

        # 过滤明显的 logo、icon、base64
        lower = src.lower()
        if lower.startswith("data:image"):
            continue
        if any(x in lower for x in ["logo", "icon", "facebook", "youtube", "line"]):
            continue

        image_urls.append(normalize_url(base_url, src))

    return unique_keep_order(image_urls)


def crop_from_text_blocks(soup: BeautifulSoup) -> str:
    """
    常见页面可能会写：
    防治作物
    水稻
    或者：
    寄主植物：水稻、玉米
    """
    crop = extract_by_label(
        soup,
        labels=[
            "防治作物",
            "寄主植物",
            "寄主植物/防治对象",
            "作物",
            "適用作物",
            "适用作物",
        ],
        max_chars=800,
    )

    if crop:
        return crop

    candidates = texts_by_selectors(
        soup,
        [
            ".crop",
            ".plant",
            ".host",
            ".tag",
            ".badge",
            ".breadcrumb li",
            "table td",
            "li",
        ],
    )

    keywords = ["水稻", "玉米", "番茄", "柑", "茶", "甘藷", "小麦", "大豆", "蔬菜", "瓜"]
    picked = []
    for text in candidates:
        if any(k in text for k in keywords) and len(text) <= 120:
            picked.append(text)

    return "；".join(unique_keep_order(picked[:5]))


def parse_detail_page(html: str, detail_url: str, base_url: str = BASE_URL) -> PestRecord:
    soup = make_soup(html)

    # 1. 病虫害名
    pest_name = text_by_selectors(
        soup,
        [
            "h1",
            "h2",
            ".title",
            ".detail-title",
            ".bug-title",
            ".pest-title",
            ".card-title",
        ],
    )

    if not pest_name:
        pest_name = extract_by_label(
            soup,
            labels=[
                "病蟲害名稱",
                "病虫害名称",
                "病蟲害名",
                "病虫害名",
                "名稱",
                "名称",
            ],
            max_chars=120,
        )

    if not pest_name:
        pest_name = extract_name_from_url(detail_url)

    # 清理标题里可能混入的站点名
    pest_name = normalize_space(
        pest_name
        .replace("農業病蟲害智能管理決策系統", "")
        .replace("农业病虫害智能管理决策系统", "")
    )

    # 2. 作物
    crop = crop_from_text_blocks(soup)
    if not crop:
        crop = extract_crop_from_url(detail_url)

    # 3. 症状摘要
    symptom_summary = extract_by_label(
        soup,
        labels=[
            "危害狀",
            "危害症狀",
            "危害症状",
            "病徵",
            "病征",
            "症狀",
            "症状",
            "發生生態",
            "发生生态",
            "危害部位",
            "危害特徵",
            "危害特征",
        ],
        max_chars=1200,
    )

    if not symptom_summary:
        # 兜底：从正文里取较长段落
        paragraphs = texts_by_selectors(
            soup,
            [
                ".content p",
                ".detail-content p",
                ".article p",
                "main p",
                "section p",
                "p",
            ],
        )
        useful = [p for p in paragraphs if 30 <= len(p) <= 800]
        symptom_summary = useful[0] if useful else ""

    # 4. 防治说明
    control_method = extract_by_label(
        soup,
        labels=[
            "防治方法",
            "防治措施",
            "防治說明",
            "防治说明",
            "管理方法",
            "綜合防治",
            "综合防治",
            "用藥資訊",
            "用药信息",
            "推薦防治",
            "推荐防治",
        ],
        max_chars=1800,
    )

    if not control_method:
        # 兜底：收集包含“防治/用药/管理/药剂”等关键词的段落
        paragraphs = texts_by_selectors(
            soup,
            [
                ".content p",
                ".detail-content p",
                ".article p",
                "main p",
                "section p",
                "p",
                "li",
                "td",
            ],
        )
        picked = []
        for p in paragraphs:
            if any(k in p for k in ["防治", "用藥", "用药", "藥劑", "药剂", "管理", "施用"]):
                picked.append(p)
        control_method = "；".join(unique_keep_order(picked[:6]))[:1800]

    # 5. 图片
    images = extract_images(soup, detail_url=detail_url, base_url=base_url)
    image_url = ";".join(images)

    record = PestRecord(
        pest_name=pest_name,
        crop=crop,
        symptom_summary=symptom_summary,
        control_method=control_method,
        image_url=image_url,
        detail_url=detail_url,
        source_site=SOURCE_SITE,
        crawled_at=now_iso(),
    )

    return record

7.3 为什么解析器要写得“宽松”

真实网站经常改版。今天详情页标题是 h1，下个月可能改成 .detail-title；今天图片写在 src，明天可能改成 data-src；今天字段叫“症状”，另一个站点可能叫“危害症状”或“病征”。

所以解析器不要只写一个选择器：

soup.select_one(".title").text

这种写法在学习阶段看起来很干净，但实际维护成本很高。更稳的写法是维护一组候选选择器和一组候选标签，谁命中就用谁。

7.4 缺失字段怎么办

缺失字段不应该导致程序崩溃。比如有些病虫害详情页没有图片，有些详情页只有防治用药信息，有些列表页标题写得不完整。

本文策略是：

  • 病虫害名优先从标题取，取不到再从标签取，再取不到从 URL query 取。
  • 作物优先从正文标签取，取不到再从 URL 的 plant 参数取。
  • 症状摘要取不到时，从正文第一段较长文本兜底。
  • 防治说明取不到时，从包含“防治、用药、管理”等关键词的段落中拼接。
  • 图片取不到时保持空字符串。
  • detail_url 永远保留，方便后续人工回查。

这样得到的数据不是百分之百完美，但不会因为一两个页面结构异常导致整个任务失败。爬虫项目的第一阶段目标是“稳定产出可回查的数据”，第二阶段才是“逐步提高字段质量”。

8️⃣ 数据存储与导出（Storage）

本文选择 CSV + JSON Lines 作为起步格式。

CSV 适合用 Excel、WPS、LibreOffice 或 pandas 直接查看。

JSON Lines 适合后续导入数据库、搜索引擎或队列系统，每行一个 JSON 对象，不怕单个字段里出现换行。

8.1 字段映射表

输出字段 Python 字段 类型 示例值
病虫害名 pest_name str 稻热病
作物 crop str 水稻
症状摘要 symptom_summary str 叶片出现病斑，严重时影响抽穗和结实
防治说明 control_method str 选用抗病品种，加强田间管理，必要时按登记说明用药
图片链接 image_url str https://example.com/images/rice_blast.jpg
详情页 detail_url str https://example.com/search/bug/detail?id=1
来源站点 source_site str 农业病虫害智能管理决策系统
抓取时间 crawled_at str 2026-06-09T10:30:00+08:00

8.2 去重策略

本项目采用两层去重。

第一层是 URL 去重。详情页 URL 相同的记录只保留一条。

第二层是内容 hash 去重。如果不同 URL 的 pest_name、crop、symptom_summary 内容完全一致，可以认为是重复内容。本文示例先使用 URL 去重，内容 hash 留作扩展。

真实项目中，我建议这样做：

  • 对 detail_url 做唯一约束。
  • 对 pest_name + crop 做辅助去重。
  • 对正文内容生成 hash，发现疑似重复时人工复核。
  • 不要只用 pest_name 去重，因为同一个病虫害可能对应多个作物。

8.3 存储层 storage.py

# crop_pest_spider/storage.py
import json
from pathlib import Path
from typing import List, Iterable

import pandas as pd

from .models import PestRecord
from .utils import logger, content_hash


FIELD_ORDER = [
    "pest_name",
    "crop",
    "symptom_summary",
    "control_method",
    "image_url",
    "detail_url",
    "source_site",
    "crawled_at",
]


CHINESE_COLUMNS = {
    "pest_name": "病虫害名",
    "crop": "作物",
    "symptom_summary": "症状摘要",
    "control_method": "防治说明",
    "image_url": "图片链接",
    "detail_url": "详情页",
    "source_site": "来源站点",
    "crawled_at": "抓取时间",
}


def dedupe_by_url(records: Iterable[PestRecord]) -> List[PestRecord]:
    seen = set()
    result = []

    for record in records:
        key = record.detail_url or content_hash(
            record.pest_name,
            record.crop,
            record.symptom_summary,
        )

        if key in seen:
            continue

        seen.add(key)
        result.append(record)

    return result


def save_jsonl(records: List[PestRecord], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)

    with path.open("w", encoding="utf-8") as f:
        for record in records:
            f.write(json.dumps(record.to_dict(), ensure_ascii=False) + "\n")

    logger.info("saved jsonl: %s, rows=%d", path, len(records))


def save_csv(records: List[PestRecord], path: Path, chinese_header: bool = True) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)

    rows = [record.to_dict() for record in records]
    df = pd.DataFrame(rows)

    for field in FIELD_ORDER:
        if field not in df.columns:
            df[field] = ""

    df = df[FIELD_ORDER]

    if chinese_header:
        df = df.rename(columns=CHINESE_COLUMNS)

    # utf-8-sig 方便 Excel 直接打开不乱码。
    df.to_csv(path, index=False, encoding="utf-8-sig")

    logger.info("saved csv: %s, rows=%d", path, len(records))

9️⃣ 运行方式与结果展示

9.1 入口文件 main.py

# crop_pest_spider/main.py
import argparse
from typing import List
from urllib.parse import quote

from tqdm import tqdm

from .config import (
    BASE_URL,
    SEARCH_URL_TEMPLATE,
    DEFAULT_KEYWORDS,
    DEFAULT_MAX_DETAILS,
    CSV_FILE,
    JSONL_FILE,
)
from .fetcher import Fetcher, FetchError
from .parser import parse_list_page, parse_detail_page
from .storage import dedupe_by_url, save_csv, save_jsonl
from .utils import logger, normalize_url


def build_search_urls(keywords: List[str]) -> List[str]:
    urls = []
    for keyword in keywords:
        encoded = quote(keyword)
        urls.append(SEARCH_URL_TEMPLATE.format(keyword=encoded))
    return urls


def collect_detail_urls(fetcher: Fetcher, keywords: List[str]) -> List[str]:
    all_detail_urls = []
    search_urls = build_search_urls(keywords)

    for search_url in search_urls:
        try:
            html = fetcher.get(search_url, referer=BASE_URL)
        except FetchError as exc:
            logger.warning("failed to fetch list page: %s", exc)
            continue

        detail_urls = parse_list_page(html, base_url=BASE_URL)
        logger.info("keyword page=%s, detail_urls=%d", search_url, len(detail_urls))

        all_detail_urls.extend(detail_urls)

    # URL 去重，保持顺序
    seen = set()
    unique_urls = []
    for url in all_detail_urls:
        url = normalize_url(BASE_URL, url)
        if url in seen:
            continue
        seen.add(url)
        unique_urls.append(url)

    return unique_urls


def crawl_details(fetcher: Fetcher, detail_urls: List[str]) -> list:
    records = []

    for detail_url in tqdm(detail_urls, desc="Crawling detail pages"):
        try:
            html = fetcher.get(detail_url, referer=BASE_URL)
        except FetchError as exc:
            logger.warning("failed to fetch detail page: %s", exc)
            continue

        if not html:
            continue

        record = parse_detail_page(html, detail_url=detail_url, base_url=BASE_URL)

        if record.is_valid():
            records.append(record)
        else:
            logger.warning("invalid record, url=%s", detail_url)

    return records


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Crop pest atlas spider: list page + detail page crawler."
    )

    parser.add_argument(
        "--keywords",
        nargs="*",
        default=DEFAULT_KEYWORDS,
        help="Search keywords, e.g. 水稻 玉米 番茄",
    )

    parser.add_argument(
        "--max-details",
        type=int,
        default=DEFAULT_MAX_DETAILS,
        help="Max detail pages to crawl. Use 0 for no limit.",
    )

    parser.add_argument(
        "--csv",
        default=str(CSV_FILE),
        help="CSV output path.",
    )

    parser.add_argument(
        "--jsonl",
        default=str(JSONL_FILE),
        help="JSONL output path.",
    )

    args = parser.parse_args()

    keywords = args.keywords
    max_details = args.max_details

    logger.info("start crawling, keywords=%s", keywords)

    fetcher = Fetcher()

    try:
        detail_urls = collect_detail_urls(fetcher, keywords)
        logger.info("collected detail urls=%d", len(detail_urls))

        if max_details and max_details > 0:
            detail_urls = detail_urls[:max_details]
            logger.info("limited detail urls=%d", len(detail_urls))

        records = crawl_details(fetcher, detail_urls)
        records = dedupe_by_url(records)

        save_csv(records, path=args.csv)
        save_jsonl(records, path=args.jsonl)

        logger.info("done, records=%d", len(records))

    finally:
        fetcher.close()


if __name__ == "__main__":
    main()

9.2 运行命令

在项目根目录执行：

python -m crop_pest_spider.main --keywords 水稻 玉米 --max-details 20

指定输出路径：

python -m crop_pest_spider.main \
  --keywords 水稻 玉米 番茄 \
  --max-details 50 \
  --csv crop_pest_spider/data/output/rice_corn_tomato_pests.csv \
  --jsonl crop_pest_spider/data/output/rice_corn_tomato_pests.jsonl

Windows PowerShell：

python -m crop_pest_spider.main --keywords 水稻 玉米 --max-details 20

9.3 输出位置

默认输出：

crop_pest_spider/data/output/crop_pest_records.csv
crop_pest_spider/data/output/crop_pest_records.jsonl

如果开启了保存原始 HTML，会看到：

crop_pest_spider/data/raw/

里面是按 URL hash 命名的 HTML 文件。解析失败时，优先打开这些文件检查页面结构。

9.4 示例结果

CSV 结果大致如下：

病虫害名 作物 症状摘要 防治说明 图片链接
稻热病 水稻 叶片、节位、穗颈等部位可能出现病斑，严重时影响植株生长和结实。 选用抗病品种，合理施肥，注意田间湿度管理，必要时按当地登记药剂说明进行防治。 https://example.com/rice_blast.jpg
白叶枯病 水稻 叶缘或叶尖出现黄白色条斑，后期病斑扩大，严重时叶片枯白。 加强种子处理和水肥管理，避免串灌漫灌，发病初期及时采取综合防治措施。 https://example.com/bacterial_leaf_blight.jpg
二化螟 水稻 幼虫蛀入茎秆，造成枯心、白穗等危害表现。 清理田间残株，保护天敌，结合虫情监测选择合适时期防治。 https://example.com/stem_borer.jpg
蚜虫 玉米 群集在叶片、嫩茎等部位吸食汁液，可能造成叶片卷曲、生长受阻。 合理密植，保护天敌，虫量较高时按推荐方法处理。 https://example.com/aphid.jpg
斜纹夜蛾 番茄 幼虫取食叶片、花和果实，严重时造成缺刻、孔洞或果面损伤。 结合诱捕、人工摘除卵块、保护天敌和适期防治进行综合管理。 https://example.com/spodoptera_litura.jpg

上面是展示格式示例。实际字段内容以目标页面解析结果为准，不建议凭空补防治建议到正式数据里。正式运行时，程序会把抓到的原文片段写入表格。

🔟 常见问题与排错

10.1 403 怎么办

403 表示服务器拒绝访问。常见原因有：

  • 请求头过于简单。
  • 访问频率太高。
  • 目标路径不希望被自动访问。
  • 页面需要登录或特定会话。
  • IP 被临时限制。

建议处理顺序：

第一，先降低频率，把 MIN_DELAY 和 MAX_DELAY 调大。

MIN_DELAY = 3
MAX_DELAY = 8

第二，检查 robots.txt 和网站声明，确认目标路径是否适合采集。

第三，补充正常 UA、Accept-Language、Referer，但不要伪造登录状态，也不要绕过限制。

第四，减少抓取规模。先抓 5 条详情页，不要一上来批量跑。

第五，如果页面明确不允许访问，就停止采集。

代理不是万能药。学习项目里不建议一遇到 403 就上代理池，这通常会把问题复杂化，也容易偏离合规边界。

10.2 429 怎么办

429 表示 Too Many Requests，请求太多了。处理方法是：

  • 降低访问频率。
  • 增加随机 sleep。
  • 减少并发。
  • 开启指数退避。
  • 分批次运行。
  • 避免重复抓同一批 URL。

本文的 tenacity 重试已经会对 429 做退避，但如果频繁出现 429，根本原因还是访问节奏太快。

可以把配置改成：

MIN_DELAY = 5
MAX_DELAY = 12
RETRY_TIMES = 2

不要为了追求速度，把对方服务器当成压测对象。爬虫写得慢一点，往往寿命更长。

10.3 HTML 抓到空壳怎么办

如果 requests 抓到的 HTML 里没有列表数据，浏览器里却能看到内容，通常说明页面是 JavaScript 动态渲染。

排查方法：

第一，保存 raw HTML，搜索病虫害名称。如果 HTML 里完全没有数据，大概率是动态渲染。

第二，打开浏览器开发者工具 Network 面板，刷新页面，查看 XHR / Fetch 请求，看看是否有 JSON 接口。

第三，如果能找到公开 JSON 接口，并且接口不需要登录、不涉及敏感信息，可以考虑请求接口。

第四，如果找不到接口，但页面公开且允许访问，可以用 Playwright 渲染页面后再解析。

下面是一个 Playwright 兜底示例：

# optional_playwright_fetch.py
from playwright.sync_api import sync_playwright


def render_html(url: str, wait_ms: int = 3000) -> str:
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/124.0 Safari/537.36"
            )
        )
        page.goto(url, wait_until="networkidle", timeout=30000)
        page.wait_for_timeout(wait_ms)
        html = page.content()
        browser.close()
        return html


if __name__ == "__main__":
    url = "https://azai.tari.gov.tw/search.html?stype=0&term=%E6%B0%B4%E7%A8%BB"
    html = render_html(url)
    print(html[:1000])

Playwright 不是用来绕过限制的，它只是浏览器自动化工具。能不用就不用，因为它更慢、更重，也更容易给服务器造成额外压力。

10.4 解析报错怎么办

解析报错通常有三类。

第一类是选择器失效。比如原来标题是 .title，现在改成 .page-title。解决方式是打开 raw HTML，重新定位字段。

第二类是字段不存在。比如某个页面没有图片，soup.select_one("img") 返回 None。解决方式是所有字段都要做空值判断。

第三类是页面结构不统一。比如病害详情和虫害详情模板不同。解决方式是按类型拆解析器，或者增加多组候选选择器。

不建议这样写：

title = soup.select_one(".title").text.strip()

更建议这样写：

node = soup.select_one(".title")
title = node.get_text(strip=True) if node else ""

10.5 编码乱码怎么办

中文网页乱码很常见。requests 会根据响应头猜编码，但不一定准确。

可以这样处理：

resp = requests.get(url, headers=headers, timeout=15)

if not resp.encoding or resp.encoding.lower() == "iso-8859-1":
    resp.encoding = resp.apparent_encoding

html = resp.text

CSV 乱码则通常是 Excel 打开方式的问题。建议导出时使用 utf-8-sig：

df.to_csv("output.csv", index=False, encoding="utf-8-sig")

这样 Windows Excel 通常能直接识别中文。

10.6 图片链接抓不到怎么办

图片抓不到的原因可能有：

  • 图片使用懒加载，真实地址在 data-src。
  • 图片是相对路径，需要 urljoin 补全。
  • 图片通过 CSS background-image 加载。
  • 图片需要浏览器渲染后才出现。
  • 页面用了图集插件，图片地址在 script 里。

本文已经兼容了常见写法：

src = (
    img.get("src")
    or img.get("data-src")
    or img.get("data-original")
    or img.get("data-lazy-src")
    or ""
)

如果图片写在 CSS 里，可以补一个正则：

import re

def extract_background_images(html: str, base_url: str) -> list:
    urls = []
    pattern = r"background-image\s*:\s*url\(['\"]?(.*?)['\"]?\)"
    for match in re.findall(pattern, html, flags=re.I):
        urls.append(urljoin(base_url, match))
    return urls

10.7 列表页没有详情链接怎么办

如果 parse_list_page 返回 0，有几个可能：

  • 页面是动态渲染，requests HTML 没有真实列表。
  • 详情链接不是 /search/bug/detail 这种模式。
  • 链接在按钮事件、script 或 JSON 里。
  • 搜索关键词没有结果。
  • 页面需要先点“直接进入首页”之类的入口。

排查顺序：

  1. 打开 data/raw 里的 HTML。
  2. 搜索关键词，比如“水稻”“稻热病”。
  3. 搜索 href=，看是否有详情链接。
  4. 搜索 detail、bug、id=。
  5. 如果 raw HTML 没数据，就用浏览器开发者工具看接口。
  6. 如果是 JS 渲染，考虑 Playwright 兜底。

1️⃣1️⃣ 进阶优化

11.1 并发抓取

当你确认目标站点允许、请求频率合适、数据量也确实较大时，可以考虑并发。但并发不是越高越好。

线程池示例：

# crop_pest_spider/concurrent_runner.py
from concurrent.futures import ThreadPoolExecutor, as_completed

from .fetcher import Fetcher, FetchError
from .parser import parse_detail_page
from .storage import dedupe_by_url, save_csv, save_jsonl
from .config import BASE_URL, CSV_FILE, JSONL_FILE
from .utils import logger


def fetch_one(detail_url: str):
    fetcher = Fetcher(min_delay=1.5, max_delay=4.0)
    try:
        html = fetcher.get(detail_url, referer=BASE_URL)
        if not html:
            return None
        return parse_detail_page(html, detail_url=detail_url, base_url=BASE_URL)
    except FetchError as exc:
        logger.warning("failed: %s, error=%s", detail_url, exc)
        return None
    finally:
        fetcher.close()


def crawl_concurrent(detail_urls: list, max_workers: int = 3):
    records = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_map = {
            executor.submit(fetch_one, url): url
            for url in detail_urls
        }

        for future in as_completed(future_map):
            url = future_map[future]
            try:
                record = future.result()
            except Exception as exc:
                logger.warning("unexpected error: %s, url=%s", exc, url)
                continue

            if record and record.is_valid():
                records.append(record)

    records = dedupe_by_url(records)
    save_csv(records, CSV_FILE)
    save_jsonl(records, JSONL_FILE)
    return records

我个人建议学习阶段 max_workers 不要超过 3。农业知识站点通常不是为高并发抓取设计的，低频慢跑更稳。

11.2 断点续跑

断点续跑的思路是保存已抓 URL。下次运行时，跳过已抓过的详情页。

# crop_pest_spider/checkpoint.py
from pathlib import Path


class Checkpoint:
    def __init__(self, path: Path):
        self.path = path
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.done = self._load()

    def _load(self) -> set:
        if not self.path.exists():
            return set()
        return set(
            line.strip()
            for line in self.path.read_text(encoding="utf-8").splitlines()
            if line.strip()
        )

    def contains(self, url: str) -> bool:
        return url in self.done

    def add(self, url: str) -> None:
        if url in self.done:
            return
        with self.path.open("a", encoding="utf-8") as f:
            f.write(url + "\n")
        self.done.add(url)

使用方式：

from pathlib import Path
from crop_pest_spider.checkpoint import Checkpoint

checkpoint = Checkpoint(Path("crop_pest_spider/data/output/done_urls.txt"))

for detail_url in detail_urls:
    if checkpoint.contains(detail_url):
        continue

    # 抓取详情页
    # ...

    checkpoint.add(detail_url)

断点续跑是非常实用的。爬虫跑到一半断网、电脑休眠、页面报错，都不至于从头再来。

11.3 日志与监控

日志至少要记录：

  • 开始时间。
  • 关键词。
  • 列表页数量。
  • 详情页数量。
  • 成功条数。
  • 失败 URL。
  • 失败原因。
  • 输出文件路径。

更进一步，可以统计成功率：

total = len(detail_urls)
success = len(records)
failed = total - success
rate = success / total * 100 if total else 0

logger.info(
    "summary: total=%d, success=%d, failed=%d, success_rate=%.2f%%",
    total,
    success,
    failed,
    rate,
)

如果后续部署到服务器，可以把日志写入文件：

import logging
from pathlib import Path

log_path = Path("crop_pest_spider/data/output/spider.log")
file_handler = logging.FileHandler(log_path, encoding="utf-8")
logger.addHandler(file_handler)

11.4 定时任务

如果你只是做一次数据整理，不需要定时任务。

如果你要定期更新，可以用 cron。

Linux crontab 示例：

crontab -e

写入：

0 3 * * 1 cd /path/to/project && /path/to/project/.venv/bin/python -m crop_pest_spider.main --keywords 水稻 玉米 --max-details 50 >> spider.log 2>&1

表示每周一凌晨 3 点运行一次。

Windows 可以用“任务计划程序”，动作里填：

程序：C:\path\to\project\.venv\Scripts\python.exe
参数：-m crop_pest_spider.main --keywords 水稻 玉米 --max-details 50
起始于：C:\path\to\project

11.5 SQLite 存储扩展

CSV 适合起步，但如果你要做增量更新，SQLite 更舒服。

# crop_pest_spider/sqlite_storage.py
import sqlite3
from pathlib import Path
from typing import List

from .models import PestRecord


CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS crop_pest_records (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    pest_name TEXT,
    crop TEXT,
    symptom_summary TEXT,
    control_method TEXT,
    image_url TEXT,
    detail_url TEXT UNIQUE,
    source_site TEXT,
    crawled_at TEXT
);
"""


INSERT_SQL = """
INSERT OR REPLACE INTO crop_pest_records (
    pest_name,
    crop,
    symptom_summary,
    control_method,
    image_url,
    detail_url,
    source_site,
    crawled_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?);
"""


def save_to_sqlite(records: List[PestRecord], db_path: Path) -> None:
    db_path.parent.mkdir(parents=True, exist_ok=True)

    conn = sqlite3.connect(db_path)
    try:
        conn.execute(CREATE_TABLE_SQL)

        rows = [
            (
                r.pest_name,
                r.crop,
                r.symptom_summary,
                r.control_method,
                r.image_url,
                r.detail_url,
                r.source_site,
                r.crawled_at,
            )
            for r in records
        ]

        conn.executemany(INSERT_SQL, rows)
        conn.commit()
    finally:
        conn.close()

查询：

import sqlite3

conn = sqlite3.connect("crop_pest_spider/data/output/crop_pest.db")

for row in conn.execute("SELECT pest_name, crop, detail_url FROM crop_pest_records LIMIT 10"):
    print(row)

conn.close()

11.6 迁移到 Scrapy

当你遇到这些需求时，可以考虑 Scrapy：

  • URL 数量很多。
  • 需要统一调度。
  • 需要下载中间件。
  • 需要自动限速。
  • 需要失败重试队列。
  • 需要更规范的数据管道。
  • 需要和数据库深度集成。

Scrapy 的结构大概是：

scrapy_project/
├── spiders/
│   └── crop_pest.py
├── items.py
├── pipelines.py
├── middlewares.py
└── settings.py

Scrapy Spider 伪代码：

import scrapy


class CropPestItem(scrapy.Item):
    pest_name = scrapy.Field()
    crop = scrapy.Field()
    symptom_summary = scrapy.Field()
    control_method = scrapy.Field()
    image_url = scrapy.Field()
    detail_url = scrapy.Field()
    source_site = scrapy.Field()
    crawled_at = scrapy.Field()


class CropPestSpider(scrapy.Spider):
    name = "crop_pest"
    allowed_domains = ["azai.tari.gov.tw"]
    start_urls = [
        "https://azai.tari.gov.tw/search.html?stype=0&term=%E6%B0%B4%E7%A8%BB"
    ]

    custom_settings = {
        "DOWNLOAD_DELAY": 3,
        "RANDOMIZE_DOWNLOAD_DELAY": True,
        "CONCURRENT_REQUESTS": 2,
        "ROBOTSTXT_OBEY": True,
        "FEED_EXPORT_ENCODING": "utf-8-sig",
    }

    def parse(self, response):
        for href in response.css("a::attr(href)").getall():
            if "/search/bug/detail" in href:
                yield response.follow(href, callback=self.parse_detail)

    def parse_detail(self, response):
        yield CropPestItem(
            pest_name=response.css("h1::text, h2::text, .title::text").get(default="").strip(),
            crop="",
            symptom_summary="",
            control_method="",
            image_url=";".join(response.css("img::attr(src)").getall()),
            detail_url=response.url,
            source_site="农业病虫害智能管理决策系统",
            crawled_at="",
        )

Scrapy 更适合规模化，但初学者最好先用 requests 把流程吃透，再迁移。框架不是捷径，它只是把成熟爬虫里常见的模块封装好了。

1️⃣2️⃣ 总结与延伸阅读

这篇文章完成了一个农作物病虫害图谱目录页的二段式采集案例。

我们从需求开始，明确要抓的字段是：

  • 病虫害名
  • 作物
  • 症状摘要
  • 防治说明
  • 图片链接

然后搭建了一个可复现的 Python 项目，拆成请求层、解析层、存储层、入口文件和工具函数。请求层处理 headers、referer、timeout、session、重试和退避；解析层处理列表页详情链接、详情页字段抽取、图片链接提取和缺失字段容错；存储层处理 CSV、JSONL、字段顺序、中文表头和 URL 去重。

这个项目最核心的价值，不在于某个选择器写得多漂亮，而在于它把爬虫工程里真正容易出问题的部分都留了位置：

  • 请求失败怎么办。
  • 页面结构变了怎么办。
  • 字段缺失怎么办。
  • 图片懒加载怎么办。
  • 列表页动态渲染怎么办。
  • 重复 URL 怎么办。
  • 数据怎么回查。
  • 怎么控制访问频率。
  • 怎么从一次性脚本升级成可维护项目。

下一步可以继续做三件事。

第一，接入 Playwright。用于处理列表页必须浏览器渲染的场景，但要保持低频访问。

第二，迁移到 Scrapy。当 URL 数量变多、任务需要稳定调度时，Scrapy 的下载器、中间件、管道和日志会更省心。

第三，做数据质量增强。比如按作物标准化名称，按病害/虫害分类，提取防治措施关键词，给图片建立本地索引，或者把详情页文本进入 Elasticsearch 做检索。

我一直觉得，爬虫最有意思的地方不只是“把网页抓下来”，而是把网页背后的知识结构整理出来。农作物病虫害图谱这个案例就很典型：网页是半结构化的，字段并不总是规整，但只要请求克制、解析稳健、存储清楚，就能把原本分散的信息变成一份可以分析、可以检索、可以复用的数据资产。

最后再强调一次：本文代码用于公开网页的数据整理学习。实际采集前，请检查目标站点规则，控制访问频率，不采集敏感信息，不绕过登录或付费限制。技术分享的边界守住了，项目才能跑得久，也更值得复用。

🌟 文末

好啦～以上就是本期的全部内容啦！如果你在实践过程中遇到任何疑问，欢迎在评论区留言交流，我看到都会尽量回复～咱们下期见！

小伙伴们在批阅的过程中，如果觉得文章不错，欢迎点赞、收藏、关注哦～
三连就是对我写作道路上最好的鼓励与支持！ ❤️🔥

✅ 专栏持续更新中｜建议收藏 + 订阅

墙裂推荐订阅专栏 👉 《Python爬虫实战》，本专栏秉承着以“入门 → 进阶 → 工程化 → 项目落地”的路线持续更新，争取让每一期内容都做到：

✅ 讲得清楚（原理）｜✅ 跑得起来（代码）｜✅ 用得上（场景）｜✅ 扛得住（工程化）

📣 想系统提升的小伙伴：强烈建议先订阅专栏 《Python爬虫实战》，再按目录大纲顺序学习，效率十倍上升～

✅ 互动征集

想让我把【某站点/某反爬/某验证码/某分布式方案】等写成某期实战？

评论区留言告诉我你的需求，我会优先安排实现(更新)哒~

⭐️ 若喜欢我，就请关注我叭～（更新不迷路）
⭐️ 若对你有用，就请点赞支持一下叭～（给我一点点动力）
⭐️ 若有疑问，就请评论留言告诉我叭～（我会补坑 & 更新迭代）

✅ 免责声明

本文爬虫思路、相关技术和代码仅用于学习参考，对阅读本文后的进行爬虫行为的用户本作者不承担任何法律责任。

使用或者参考本项目即表示您已阅读并同意以下条款：

  • 合法使用： 不得将本项目用于任何违法、违规或侵犯他人权益的行为，包括但不限于网络攻击、诈骗、绕过身份验证、未经授权的数据抓取等。
  • 风险自负： 任何因使用本项目而产生的法律责任、技术风险或经济损失，由使用者自行承担，项目作者不承担任何形式的责任。
  • 禁止滥用： 不得将本项目用于违法牟利、黑产活动或其他不当商业用途。
  • 使用或者参考本项目即视为同意上述条款,即 “谁使用，谁负责” 。如不同意，请立即停止使用并删除本项目。！！！
# 爬虫 # python
Logo
亚马逊云科技技术品牌专区

更多推荐

2026年量化工具选择，要跟着能力基础走

读者应理解，选择工具前要先判断自己的能力基础和当前任务。工具应该帮助自己补上当前最关键的缺口，而不是替代学习顺序本身。

avatar 亚马逊云科技技术品牌专区

近期零基础学量化，先按基础差异拆学习顺序

读者应明白，拆解学习顺序时要先看自己的基础缺口，再决定哪一段需要放慢。不同流程阶段都有不同检查任务，学习路线要能反映这些差异。

avatar 亚马逊云科技技术品牌专区

加州理工 CS367 C 语言系统编程笔记（一）

C语言是一种强大且广泛使用的编程语言，尤其在系统编程领域。本节课我们将学习C语言的基础语法，包括如何编写“Hello, World!”程序、声明变量以及使用基本数据类型。我们将通过实际的代码示例来加深理解。本节课中我们一起学习了C语言的基础语法，包括如何编写和运行一个简单的C程序、声明变量以及使用基本数据类型。我们还介绍了如何使用Shell环境来编译和运行程序。通过对比Java，你可以看到C语言在

avatar 亚马逊云科技技术品牌专区