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

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

这篇文章要做的事情很明确：用 Python 编写一个面向“科普馆展项目录分页”的爬虫，从目录页逐页采集展项链接，再进入详情页解析出展项名、学科主题、适龄范围、互动方式和链接，最终导出为 CSV 文件。

读完这篇文章，你可以获得：

1. 一套“目录页 + 详情页”结构化采集的完整实现思路；
2. 一个可运行、可改造、带容错和去重逻辑的 Python 爬虫项目；
3. 面对分页、字段缺失、403/429、动态渲染等问题时的排查方法。

我个人比较喜欢这种类型的爬虫练习，因为它既不像简单列表页那样单薄，也不像大型平台采集那样重。它刚好覆盖了日常数据采集里最常见的一组问题：分页、详情链接、字段清洗、失败重试、导出和去重。把这一套吃透，再去写展会目录、课程目录、博物馆藏品目录、公开活动目录，本质上都能举一反三。

1️⃣ 摘要（Abstract）

本文将使用 Python、requests、BeautifulSoup 和 lxml，实现一个科普馆展项目录分页爬虫：先从目录页提取详情页链接，再进入详情页提取展项名、学科主题、适龄范围、互动方式和页面链接，最后保存为 CSV 文件。

读完本文后，你能够掌握：

1. 目录页与详情页分层采集的工程化写法；
2. 请求层、解析层、存储层拆分的基本设计；
3. 如何为真实网站改造选择器、分页逻辑和容错策略。

本文示例不绑定某一家具体科普馆网站，因为真实网站经常改版，直接写死某个站点反而不利于复用。代码会提供一个默认 demo 模式，可以直接运行看到结果；如果你要采集真实公开网页，只需要把目录页 URL 模板和 CSS 选择器替换成目标网站对应结构即可。

2️⃣ 背景与需求（Why）

科普馆、科技馆、博物馆、展览中心这类网站，通常会公开展示展项或展览内容。一个典型页面结构是：

• 目录页展示多个展项卡片；
• 每个卡片包含展项标题、缩略图、简介或“查看详情”链接；
• 目录页支持分页；
• 详情页展示完整信息，例如适合年龄、学科主题、互动方式、展项说明等。

如果只是人工浏览，几十个展项还能勉强处理。但一旦需要做汇总表、横向对比、主题统计、适龄分布分析，手工复制就会非常低效，而且容易出错。

为什么要爬这类数据？

常见需求有三类。

第一类是数据分析。比如你想知道某科普馆的展项主要集中在哪些学科主题，是物理更多，还是生命科学更多；适龄范围是否偏向儿童，还是覆盖青少年和成人；互动方式是动手实验多，还是多媒体观看多。

第二类是信息聚合。多个科普馆的公开展项信息分散在不同网页上，统一采集后可以建立一个可查询的小型资料库，方便检索和筛选。

第三类是自动化维护。假设你要定期检查展项是否新增、下架或更新介绍，用爬虫定时跑一遍，结合 URL 去重和内容 hash，就能快速发现变化。

目标字段

本文目标字段如下：

字段	说明
展项名	展项或展品的名称
学科主题	如物理、生命科学、地球科学、信息技术等
适龄范围	如 6 岁以上、8–12 岁、亲子家庭、青少年等
互动方式	如动手实验、触摸屏互动、沉浸式体验、观察演示等
链接	详情页 URL，用于追溯来源

这五个字段比较适合入门，因为它们既有文本字段，也有分类字段，还包含一个天然去重键：链接。

---

3️⃣ 合规与注意事项

爬虫不是“能抓就抓”，更不是“越快越好”。一个专业的爬虫项目，第一步永远不是写代码，而是判断是否适合采集、如何低影响地采集。

robots.txt 基本说明

很多网站会在根目录提供 robots.txt 文件，例如：

https://example.com/robots.txt

robots.txt 用来告诉搜索引擎或自动化程序，哪些路径允许访问，哪些路径不建议访问。它不是法律文本，但它是网站对自动访问行为的基本声明。写爬虫前，建议先查看目标站点的 robots.txt，确认目录页和详情页是否允许抓取。

例如你可以手动访问：

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

重点关注：

User-agent: *
Disallow: /admin/
Disallow: /login/
Allow: /exhibits/

如果目标路径明确不允许访问，就不要采集。如果没有 robots.txt，也不代表可以无节制抓取，仍然要控制频率。

频率控制

科普馆网站通常不是大型商业平台，很多是内容展示型站点，服务器性能未必很强。爬虫访问频率应该温和一些。

建议：

• 每次请求之间 sleep 0.5–2 秒；
• 不使用攻击式并发；
• 遇到 429 Too Many Requests 主动降速；
• 遇到 5xx 错误不要连续猛刷；
• 对失败请求做有限次数重试，而不是无限循环。

本文代码中会加入请求间隔和重试退避逻辑。

不采集敏感信息

本文只采集公开展项内容，不涉及用户数据、个人信息、登录后页面、付费内容或后台接口。

实际开发时建议坚持几个原则：

• 不绕过登录限制；
• 不绕过付费限制；
• 不采集个人敏感信息；
• 不破解验证码；
• 不模拟破坏性请求；
• 不对网站造成明显访问压力。

这类技术分享的重点是结构化公开信息，而不是挑战网站防护。

---

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

静态、动态还是 API？

爬虫选型前，先判断页面数据从哪里来。

常见情况有三种。

第一种是静态 HTML。网页源码里已经包含展项卡片和详情信息。这种情况最适合用 requests + BeautifulSoup 或 lxml。

第二种是动态渲染。网页源码只有一个空壳，真正的数据由 JavaScript 加载。此时可以继续分析接口，也可以使用 Playwright 渲染页面。

第三种是公开 API。页面数据来自 JSON 接口，比如 /api/exhibits?page=1。这种情况下，优先抓接口而不是解析 HTML。

本文主线按“静态 HTML 目录页 + 静态 HTML 详情页”设计。理由是科普馆、博物馆、展览馆这类内容展示站点，经常采用服务端渲染页面，HTML 结构比较稳定，用 requests 和 BeautifulSoup 更轻、更容易部署。

整体流程

流程可以概括为：

采集目录页
    ↓
解析展项详情链接
    ↓
请求详情页
    ↓
解析字段
    ↓
清洗文本
    ↓
去重
    ↓
写入 CSV

换成更工程化的表达，就是：

Fetcher 请求层
    负责 headers、timeout、重试、退避、session

Parser 解析层
    负责目录页链接提取、详情页字段抽取、字段缺失容错

Storage 存储层
    负责 CSV 写入、URL 去重、字段映射

Runner 调度层
    负责分页循环、请求节奏、日志输出

我建议初学者从 requests + bs4 开始，而不是一上来就 Scrapy。Scrapy 很强，但它有自己的项目结构、调度机制、Item Pipeline 和中间件体系。对于一篇实战教程来说，先把“请求—解析—存储”的核心逻辑写清楚，理解成本更低。

当数据量变大，或者需要并发、断点续跑、失败队列、分布式调度时，再迁移到 Scrapy 会更自然。

---

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

Python 版本

建议使用 Python 3.10 或 Python 3.11。本文代码在 Python 3.11 思路下编写，Python 3.10 也可以运行。

查看版本：

python --version

创建虚拟环境

mkdir science_exhibit_spider
cd science_exhibit_spider

python -m venv .venv

Windows 激活：

.venv\Scripts\activate

macOS / Linux 激活：

source .venv/bin/activate

安装依赖

pip install requests beautifulsoup4 lxml

生成 requirements.txt：

requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.3.0

安装时：

pip install -r requirements.txt

推荐项目结构

science_exhibit_spider/
├── run.py
├── requirements.txt
├── config.py
├── spider/
│   ├── __init__.py
│   ├── fetcher.py
│   ├── parser.py
│   └── storage.py
├── data/
│   └── exhibits.csv
└── logs/
    └── spider.log

这个结构不复杂，但足够清晰。请求、解析、存储分开后，后续维护会舒服很多。很多爬虫项目后期失控，往往不是因为技术难，而是因为一开始把所有逻辑塞在一个脚本里，几百行之后自己也不愿意再改。

---

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

请求层负责和目标网站打交道。它应该关心：

• headers；
• User-Agent；
• Referer；
• timeout；
• session；
• cookie；
• 失败重试；
• 退避等待；
• 状态码处理。

config.py

先写配置文件。

# config.py
from dataclasses import dataclass, field
from typing import Dict, List


@dataclass
class SpiderConfig:
    """
    科普馆展项目录爬虫配置。

    默认使用 demo.local 作为演示域名。
    如果采集真实网站，把 demo_mode 改为 False，
    并传入真实 LIST_URL_TEMPLATE 和选择器即可。
    """

    demo_mode: bool = True

    # 真实网站示例：
    # list_url_template: str = "https://www.example.com/exhibits?page={page}"
    list_url_template: str = "https://demo.local/exhibits?page={page}"

    start_page: int = 1
    max_pages: int = 5

    request_timeout: int = 12
    max_retries: int = 3
    backoff_base: float = 1.5

    list_delay: float = 1.0
    detail_delay: float = 0.5

    user_agent: str = (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0 Safari/537.36"
    )

    referer: str = "https://demo.local/"

    # 列表页选择器
    list_item_selector: str = ".exhibit-card"
    detail_link_selector: str = "a.detail-link::attr(href)"
    next_page_selector: str = "a.next-page::attr(href)"

    # 详情页字段选择器
    detail_selectors: Dict[str, str] = field(
        default_factory=lambda: {
            "展项名": "h1.exhibit-title",
            "学科主题": ".field-subject",
            "适龄范围": ".field-age",
            "互动方式": ".field-interaction",
        }
    )

    output_fields: List[str] = field(
        default_factory=lambda: [
            "展项名",
            "学科主题",
            "适龄范围",
            "互动方式",
            "链接",
        ]
    )

这里使用 dataclass，是因为配置项较多，用普通字典容易写错字段名。dataclass 既清楚，又方便 IDE 提示。

spider/fetcher.py

下面实现请求层。

# spider/fetcher.py
import logging
import random
import time
from typing import Optional
from urllib.parse import urlparse, parse_qs

import requests


class FetchError(Exception):
    """请求失败时抛出的自定义异常。"""


class HttpFetcher:
    """
    HTTP 请求层。

    功能：
    1. 复用 requests.Session；
    2. 设置 headers；
    3. 设置 timeout；
    4. 对常见失败状态码做有限重试；
    5. 对 429 支持 Retry-After；
    6. 使用指数退避降低访问压力。
    """

    def __init__(self, config):
        self.config = config
        self.session = requests.Session()
        self.logger = logging.getLogger(self.__class__.__name__)

        self.session.headers.update(
            {
                "User-Agent": config.user_agent,
                "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,en;q=0.8",
                "Connection": "keep-alive",
                "Referer": config.referer,
            }
        )

    def fetch(self, url: str) -> str:
        last_error: Optional[Exception] = None

        for attempt in range(1, self.config.max_retries + 1):
            try:
                self.logger.info("Fetching %s, attempt=%s", url, attempt)

                response = self.session.get(
                    url,
                    timeout=self.config.request_timeout,
                )

                if response.status_code == 429:
                    wait_seconds = self._get_retry_after(response)
                    self.logger.warning(
                        "Got 429 for %s, sleep %.2f seconds",
                        url,
                        wait_seconds,
                    )
                    time.sleep(wait_seconds)
                    continue

                if response.status_code in {500, 502, 503, 504}:
                    raise FetchError(
                        f"Server error {response.status_code}: {url}"
                    )

                if response.status_code in {401, 403}:
                    raise FetchError(
                        f"Access denied {response.status_code}: {url}"
                    )

                response.raise_for_status()

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

                return response.text

            except Exception as exc:
                last_error = exc

                if attempt >= self.config.max_retries:
                    break

                sleep_seconds = self._backoff_sleep(attempt)
                self.logger.warning(
                    "Fetch failed: %s, sleep %.2f seconds before retry",
                    exc,
                    sleep_seconds,
                )
                time.sleep(sleep_seconds)

        raise FetchError(f"Failed to fetch {url}: {last_error}")

    def _get_retry_after(self, response: requests.Response) -> float:
        retry_after = response.headers.get("Retry-After")
        if retry_after and retry_after.isdigit():
            return float(retry_after)

        return self._backoff_sleep(1) + random.uniform(0.5, 1.5)

    def _backoff_sleep(self, attempt: int) -> float:
        return self.config.backoff_base ** attempt + random.uniform(0.2, 0.8)


class DemoFetcher:
    """
    演示请求层。

    为了让本文代码在没有真实目标网站的情况下也能直接运行，
    这里内置了几页模拟 HTML。实际项目中使用 HttpFetcher 即可。
    """

    def __init__(self, config):
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)

    def fetch(self, url: str) -> str:
        parsed = urlparse(url)

        if parsed.netloc != "demo.local":
            raise FetchError(f"DemoFetcher only supports demo.local: {url}")

        if parsed.path == "/exhibits":
            query = parse_qs(parsed.query)
            page = int(query.get("page", ["1"])[0])
            html = self._list_page(page)
            if html:
                return html
            raise FetchError(f"Demo list page not found: {url}")

        if parsed.path.startswith("/exhibits/"):
            exhibit_id = parsed.path.rstrip("/").split("/")[-1]
            html = self._detail_page(exhibit_id)
            if html:
                return html
            raise FetchError(f"Demo detail page not found: {url}")

        raise FetchError(f"Demo URL not found: {url}")

    def _list_page(self, page: int) -> str:
        pages = {
            1: """
            <html>
              <body>
                <section class="exhibit-list">
                  <article class="exhibit-card">
                    <h2>风洞实验台</h2>
                    <a class="detail-link" href="/exhibits/001">查看详情</a>
                  </article>
                  <article class="exhibit-card">
                    <h2>人体导电球</h2>
                    <a class="detail-link" href="/exhibits/002">查看详情</a>
                  </article>
                  <article class="exhibit-card">
                    <h2>地震模拟平台</h2>
                    <a class="detail-link" href="/exhibits/003">查看详情</a>
                  </article>
                </section>
                <nav class="pagination">
                  <a class="next-page" href="/exhibits?page=2">下一页</a>
                </nav>
              </body>
            </html>
            """,
            2: """
            <html>
              <body>
                <section class="exhibit-list">
                  <article class="exhibit-card">
                    <h2>太阳系漫游</h2>
                    <a class="detail-link" href="/exhibits/004">查看详情</a>
                  </article>
                  <article class="exhibit-card">
                    <h2>声音可视化</h2>
                    <a class="detail-link" href="/exhibits/005">查看详情</a>
                  </article>
                </section>
                <nav class="pagination">
                </nav>
              </body>
            </html>
            """,
        }
        return pages.get(page, "")

    def _detail_page(self, exhibit_id: str) -> str:
        details = {
            "001": """
            <html>
              <body>
                <main class="detail">
                  <h1 class="exhibit-title">风洞实验台</h1>
                  <div class="meta">
                    <p>学科主题：<span class="field-subject">物理 / 空气动力学</span></p>
                    <p>适龄范围：<span class="field-age">8 岁以上</span></p>
                    <p>互动方式：<span class="field-interaction">动手实验、观察演示</span></p>
                  </div>
                </main>
              </body>
            </html>
            """,
            "002": """
            <html>
              <body>
                <main class="detail">
                  <h1 class="exhibit-title">人体导电球</h1>
                  <div class="meta">
                    <p>学科主题：<span class="field-subject">物理 / 电学</span></p>
                    <p>适龄范围：<span class="field-age">6 岁以上</span></p>
                    <p>互动方式：<span class="field-interaction">触摸互动、现象观察</span></p>
                  </div>
                </main>
              </body>
            </html>
            """,
            "003": """
            <html>
              <body>
                <main class="detail">
                  <h1 class="exhibit-title">地震模拟平台</h1>
                  <div class="meta">
                    <p>学科主题：<span class="field-subject">地球科学 / 防灾减灾</span></p>
                    <p>适龄范围：<span class="field-age">10 岁以上</span></p>
                    <p>互动方式：<span class="field-interaction">沉浸体验、情景模拟</span></p>
                  </div>
                </main>
              </body>
            </html>
            """,
            "004": """
            <html>
              <body>
                <main class="detail">
                  <h1 class="exhibit-title">太阳系漫游</h1>
                  <div class="meta">
                    <p>学科主题：<span class="field-subject">天文学 / 空间科学</span></p>
                    <p>适龄范围：<span class="field-age">7 岁以上</span></p>
                    <p>互动方式：<span class="field-interaction">多媒体互动、模型观察</span></p>
                  </div>
                </main>
              </body>
            </html>
            """,
            "005": """
            <html>
              <body>
                <main class="detail">
                  <h1 class="exhibit-title">声音可视化</h1>
                  <div class="meta">
                    <p>学科主题：<span class="field-subject">物理 / 声学</span></p>
                    <p>适龄范围：<span class="field-age">6–12 岁</span></p>
                    <p>互动方式：<span class="field-interaction">声音实验、屏幕反馈</span></p>
                  </div>
                </main>
              </body>
            </html>
            """,
        }
        return details.get(exhibit_id, "")

请求层有几个细节值得说。

第一，使用 requests.Session()。它可以复用连接，也方便统一设置 headers。如果目标站点需要普通 cookie，比如首次访问首页后再访问列表页，session 也可以保留 cookie。

第二，设置 timeout。没有 timeout 的请求是很危险的，网络卡住时程序可能一直挂着。爬虫里我几乎都会给请求加 timeout。

第三，重试要有限制。重试不是硬怼服务器，而是处理临时网络波动。超过最大重试次数后，应该记录失败并跳过，不能无限循环。

第四，429 要降速。429 通常表示请求过快。遇到这种状态码，不要立刻继续访问。

---

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

解析层负责把 HTML 转成结构化数据。本文使用 BeautifulSoup，底层解析器用 lxml。

解析层要解决四个问题：

1. 目录页如何拿到详情链接；
2. 下一页链接如何获取；
3. 详情页字段如何抽取；
4. 缺失字段如何处理。

spider/parser.py

# spider/parser.py
import logging
import re
from typing import Dict, List, Optional, Tuple
from urllib.parse import urljoin

from bs4 import BeautifulSoup, Tag


class ExhibitParser:
    """
    科普馆展项解析器。

    支持普通 CSS 选择器，也支持两种简化写法：
    1. a.detail::attr(href)
    2. h1.title::text

    BeautifulSoup 本身不直接支持 ::attr() 这种语法，
    这里做一层轻量封装，是为了让配置文件更直观。
    """

    def __init__(self, config):
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)

    def parse_list(self, html: str, page_url: str) -> Tuple[List[str], Optional[str]]:
        soup = BeautifulSoup(html, "lxml")

        cards = soup.select(self.config.list_item_selector)
        detail_urls: List[str] = []

        for card in cards:
            href = self.extract_value(card, self.config.detail_link_selector)

            if not href:
                self.logger.warning("Detail link missing on page %s", page_url)
                continue

            detail_urls.append(urljoin(page_url, href))

        next_href = self.extract_value(soup, self.config.next_page_selector)
        next_url = urljoin(page_url, next_href) if next_href else None

        return detail_urls, next_url

    def parse_detail(self, html: str, detail_url: str) -> Dict[str, str]:
        soup = BeautifulSoup(html, "lxml")

        item: Dict[str, str] = {}

        for field_name, selector in self.config.detail_selectors.items():
            raw_value = self.extract_value(soup, selector)
            item[field_name] = self.clean_text(raw_value)

        item["链接"] = detail_url

        # 容错：展项名为空时，用 URL 尾部做一个弱兜底。
        # 真实项目里也可以把这个记录到日志，后续人工复核。
        if not item.get("展项名"):
            item["展项名"] = f"未命名展项-{detail_url.rstrip('/').split('/')[-1]}"

        for field in self.config.output_fields:
            item.setdefault(field, "")

        return item

    def extract_value(self, root, selector: str) -> str:
        css_selector, mode, attr_name = self._parse_selector(selector)

        node = root.select_one(css_selector)
        if not node:
            return ""

        if mode == "attr":
            if isinstance(node, Tag):
                return str(node.get(attr_name, "")).strip()
            return ""

        return node.get_text(" ", strip=True)

    def _parse_selector(self, selector: str):
        selector = selector.strip()

        attr_match = re.search(r"::attr\((.*?)\)$", selector)
        if attr_match:
            attr_name = attr_match.group(1).strip()
            css_selector = selector[: attr_match.start()].strip()
            return css_selector, "attr", attr_name

        if selector.endswith("::text"):
            css_selector = selector[: -len("::text")].strip()
            return css_selector, "text", None

        return selector, "text", None

    def clean_text(self, value: str) -> str:
        if value is None:
            return ""

        value = value.replace("\xa0", " ")
        value = re.sub(r"\s+", " ", value)
        value = value.strip(" \t\r\n：:")

        return value

这里的解析器故意没有写得太花。对这类站点来说，最容易出问题的不是算法，而是 HTML 结构变化。因此解析层应该尽量做到：

• 选择器集中配置；
• 字段缺失不直接崩溃；
• 解析异常可定位；
• 输出字段稳定。

列表页如何拿详情链接？

列表页通常有这样的结构：

<article class="exhibit-card">
  <h2>风洞实验台</h2>
  <a class="detail-link" href="/exhibits/001">查看详情</a>
</article>

我们先选中每个卡片：

cards = soup.select(".exhibit-card")

再从卡片内部提取详情链接：

href = card.select_one("a.detail-link").get("href")

最后用 urljoin 转成绝对 URL：

detail_url = urljoin(page_url, href)

为什么要用 urljoin？因为真实网站里详情链接经常是相对路径，比如：

/exhibits/001

也可能是：

../detail/001.html

不用 urljoin 的话，后续请求很容易拼错。

详情页如何抽字段？

详情页结构可能类似：

<h1 class="exhibit-title">风洞实验台</h1>
<p>学科主题：<span class="field-subject">物理 / 空气动力学</span></p>
<p>适龄范围：<span class="field-age">8 岁以上</span></p>
<p>互动方式：<span class="field-interaction">动手实验、观察演示</span></p>

对应选择器为：

{
    "展项名": "h1.exhibit-title",
    "学科主题": ".field-subject",
    "适龄范围": ".field-age",
    "互动方式": ".field-interaction",
}

真实项目里，如果详情页字段是表格，可以改成：

{
    "展项名": "h1",
    "学科主题": "table.info tr:nth-child(2) td:nth-child(2)",
    "适龄范围": "table.info tr:nth-child(3) td:nth-child(2)",
    "互动方式": "table.info tr:nth-child(4) td:nth-child(2)",
}

如果字段是 JSON-LD，也可以从 script 标签里读取 JSON。解析方式不固定，关键是把“字段名”和“选择器”分离出来。

缺失字段怎么办？

字段缺失是正常情况，不是异常情况。比如有些展项详情页可能没有写适龄范围，有些只写了简介，没有单独标互动方式。

本文策略是：

• 选择器找不到时返回空字符串；
• 展项名缺失时用 URL 尾部做临时兜底；
• 所有输出字段都保留；
• 后续可以人工复核空字段。

这样导出的 CSV 结构是稳定的，不会因为某一条数据缺字段导致整批数据失败。

---

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

对于入门项目，CSV 是最合适的存储格式。它简单、直观，可以直接用 Excel、WPS、Numbers 或 pandas 打开。

如果后续数据量变大，再考虑 SQLite、MySQL 或 PostgreSQL。

字段映射表

CSV 字段名	类型	示例值
展项名	str	风洞实验台
学科主题	str	物理 / 空气动力学
适龄范围	str	8 岁以上
互动方式	str	动手实验、观察演示
链接	str	https://demo.local/exhibits/001

去重策略

本文采用 URL 唯一去重。原因很简单：详情页链接通常是展项最稳定的唯一标识。

实际项目可以进一步增强：

• URL 唯一；
• 展项名 + 学科主题组合去重；
• 详情正文 hash 去重；
• 最近更新时间判断。

如果网站经常改 URL，但内容不变，可以考虑内容 hash。如果网站 URL 稳定，用 URL 去重就足够。

spider/storage.py

# spider/storage.py
import csv
from pathlib import Path
from typing import Dict, Iterable, Set


class CsvStorage:
    """
    CSV 存储层。

    功能：
    1. 自动创建输出目录；
    2. 首次写入时创建表头；
    3. 根据 URL 加载已有数据，实现增量去重；
    4. 逐条追加，降低内存压力。
    """

    def __init__(self, output_path: str, fields: Iterable[str]):
        self.output_path = Path(output_path)
        self.fields = list(fields)
        self.output_path.parent.mkdir(parents=True, exist_ok=True)

    def load_seen_urls(self) -> Set[str]:
        seen: Set[str] = set()

        if not self.output_path.exists():
            return seen

        with self.output_path.open("r", encoding="utf-8-sig", newline="") as f:
            reader = csv.DictReader(f)
            for row in reader:
                url = row.get("链接")
                if url:
                    seen.add(url)

        return seen

    def save_item(self, item: Dict[str, str]) -> None:
        file_exists = self.output_path.exists()

        with self.output_path.open("a", encoding="utf-8-sig", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=self.fields)

            if not file_exists:
                writer.writeheader()

            safe_item = {field: item.get(field, "") for field in self.fields}
            writer.writerow(safe_item)

这里编码使用 utf-8-sig，主要是为了让 Excel 打开 CSV 时更少出现中文乱码。如果你主要用 pandas 处理，也可以直接用 utf-8。

---

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

现在写入口文件 run.py。

run.py

# run.py
import argparse
import logging
import sys
import time
from pathlib import Path
from typing import Optional

from config import SpiderConfig
from spider.fetcher import DemoFetcher, FetchError, HttpFetcher
from spider.parser import ExhibitParser
from spider.storage import CsvStorage


def setup_logging() -> None:
    Path("logs").mkdir(exist_ok=True)

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
        handlers=[
            logging.StreamHandler(sys.stdout),
            logging.FileHandler("logs/spider.log", encoding="utf-8"),
        ],
    )


def build_config(args: argparse.Namespace) -> SpiderConfig:
    config = SpiderConfig()

    config.demo_mode = args.demo
    config.max_pages = args.max_pages
    config.start_page = args.start_page

    if args.list_url_template:
        config.list_url_template = args.list_url_template

    if args.list_item_selector:
        config.list_item_selector = args.list_item_selector

    if args.detail_link_selector:
        config.detail_link_selector = args.detail_link_selector

    if args.next_page_selector:
        config.next_page_selector = args.next_page_selector

    return config


def crawl(config: SpiderConfig, output_path: str) -> None:
    logger = logging.getLogger("Runner")

    fetcher = DemoFetcher(config) if config.demo_mode else HttpFetcher(config)
    parser = ExhibitParser(config)
    storage = CsvStorage(output_path, config.output_fields)

    seen_urls = storage.load_seen_urls()

    logger.info("Seen URL count: %s", len(seen_urls))

    page = config.start_page
    current_url: Optional[str] = config.list_url_template.format(page=page)

    total_saved = 0
    total_failed = 0

    while current_url and page <= config.max_pages:
        logger.info("Start list page: %s", current_url)

        try:
            list_html = fetcher.fetch(current_url)
        except FetchError as exc:
            logger.error("List page failed: %s", exc)
            break

        detail_urls, next_url = parser.parse_list(list_html, current_url)

        if not detail_urls:
            logger.warning("No detail URLs found on %s", current_url)
            break

        logger.info(
            "Page %s detail URL count: %s",
            page,
            len(detail_urls),
        )

        for detail_url in detail_urls:
            if detail_url in seen_urls:
                logger.info("Skip duplicated URL: %s", detail_url)
                continue

            try:
                detail_html = fetcher.fetch(detail_url)
                item = parser.parse_detail(detail_html, detail_url)

                storage.save_item(item)
                seen_urls.add(detail_url)
                total_saved += 1

                logger.info(
                    "Saved: %s | %s",
                    item.get("展项名", ""),
                    detail_url,
                )

            except Exception as exc:
                total_failed += 1
                logger.exception("Detail page failed: %s, error=%s", detail_url, exc)

            time.sleep(config.detail_delay)

        if next_url:
            current_url = next_url
        else:
            page += 1
            if page > config.max_pages:
                break
            current_url = config.list_url_template.format(page=page)

        time.sleep(config.list_delay)

    logger.info(
        "Finished. saved=%s, failed=%s, output=%s",
        total_saved,
        total_failed,
        output_path,
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Crawl science museum exhibit catalog pages and export CSV."
    )

    parser.add_argument(
        "--demo",
        action="store_true",
        help="Run with built-in demo HTML pages.",
    )

    parser.add_argument(
        "--list-url-template",
        default="",
        help=(
            "List page URL template, e.g. "
            "'https://www.example.com/exhibits?page={page}'"
        ),
    )

    parser.add_argument(
        "--start-page",
        type=int,
        default=1,
        help="Start page number.",
    )

    parser.add_argument(
        "--max-pages",
        type=int,
        default=5,
        help="Max list pages to crawl.",
    )

    parser.add_argument(
        "--output",
        default="data/exhibits.csv",
        help="Output CSV path.",
    )

    parser.add_argument(
        "--list-item-selector",
        default="",
        help="CSS selector for each item on list page.",
    )

    parser.add_argument(
        "--detail-link-selector",
        default="",
        help="CSS selector for detail link. Support ::attr(href).",
    )

    parser.add_argument(
        "--next-page-selector",
        default="",
        help="CSS selector for next page link. Support ::attr(href).",
    )

    return parser.parse_args()


def main() -> None:
    setup_logging()
    args = parse_args()

    if not args.demo and not args.list_url_template:
        raise SystemExit(
            "Online mode requires --list-url-template. "
            "For quick test, run: python run.py --demo"
        )

    config = build_config(args)
    crawl(config, args.output)


if __name__ == "__main__":
    main()

创建 spider/init.py

# spider/__init__.py

空文件即可，表示 spider 是一个 Python 包。

运行 demo

python run.py --demo

输出文件：

data/exhibits.csv

日志文件：

logs/spider.log

demo 输出示例

CSV 内容类似：

展项名,学科主题,适龄范围,互动方式,链接
风洞实验台,物理 / 空气动力学,8 岁以上,动手实验、观察演示,https://demo.local/exhibits/001
人体导电球,物理 / 电学,6 岁以上,触摸互动、现象观察,https://demo.local/exhibits/002
地震模拟平台,地球科学 / 防灾减灾,10 岁以上,沉浸体验、情景模拟,https://demo.local/exhibits/003
太阳系漫游,天文学 / 空间科学,7 岁以上,多媒体互动、模型观察,https://demo.local/exhibits/004
声音可视化,物理 / 声学,6–12 岁,声音实验、屏幕反馈,https://demo.local/exhibits/005

对接真实网站

假设真实目录页是：

https://www.example-museum.org/exhibits?page=1
https://www.example-museum.org/exhibits?page=2

并且列表页结构是：

<div class="item">
  <a class="more" href="/exhibit/detail/123">详情</a>
</div>

那么可以这样运行：

python run.py \
  --list-url-template "https://www.example-museum.org/exhibits?page={page}" \
  --list-item-selector ".item" \
  --detail-link-selector "a.more::attr(href)" \
  --next-page-selector ".pagination .next::attr(href)" \
  --max-pages 10 \
  --output data/exhibits.csv

如果详情页字段选择器不同，需要改 config.py 里的 detail_selectors。比如：

detail_selectors={
    "展项名": ".detail-title",
    "学科主题": ".info .subject",
    "适龄范围": ".info .age",
    "互动方式": ".info .interaction",
}

如果网站没有“下一页”按钮，而是纯页码分页，本文代码也可以工作，因为它会在没有 next_url 时使用 list_url_template 继续生成下一页 URL。

---

🔟 常见问题与排错

1. 遇到 403 怎么办？

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

• User-Agent 过于简单；
• Referer 缺失；
• 目标路径不允许自动访问；
• 站点设置了访问限制；
• 访问频率异常。

排查顺序建议如下：

第一，确认 URL 在浏览器中能正常打开。

第二，查看 robots.txt，确认目标路径是否适合采集。

第三，设置合理 headers，不要使用明显异常的 User-Agent。

第四，降低访问频率。

第五，如果仍然 403，就应该停止采集或联系网站方确认授权，而不是尝试绕过限制。

2. 遇到 429 怎么办？

429 通常表示请求太频繁。处理方式很简单：降速。

本文代码已经对 429 做了处理，会读取 Retry-After，如果服务器没有返回该字段，就使用退避等待。

还可以继续优化：

config.list_delay = 2.0
config.detail_delay = 1.5
config.max_retries = 2

爬虫越稳定，越应该像一个有耐心的普通访问者，而不是压测工具。

3. HTML 抓到空壳怎么办？

如果你打印 response.text，发现里面只有：

<div id="app"></div>
<script src="/static/app.js"></script>

说明页面可能是前端动态渲染。

这时有两个方向。

第一，打开浏览器开发者工具，查看 Network 面板，寻找返回 JSON 的接口。如果能找到公开接口，优先请求接口。

第二，如果数据必须渲染后才出现，可以用 Playwright：

pip install playwright
playwright install

简单示例：

from playwright.sync_api import sync_playwright


def fetch_rendered_html(url: str) -> 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)
        html = page.content()
        browser.close()
        return html

不过 Playwright 更重，速度更慢，部署也更复杂。能抓静态 HTML 或接口时，不建议一上来就使用浏览器渲染。

4. 解析报错怎么办？

常见解析错误有三类。

第一，选择器写错。可以把 HTML 保存下来，用浏览器或脚本单独测试选择器。

第二，页面结构不一致。有些详情页字段完整，有些字段缺失。解析层要允许字段为空。

第三，网站改版。选择器失效后，不要急着改业务逻辑，先重新观察 HTML 结构，再更新配置。

调试时可以临时加几行代码：

from pathlib import Path

Path("debug.html").write_text(html, encoding="utf-8")

然后用浏览器打开 debug.html，看实际抓到的内容到底是什么。

5. 编码乱码怎么办？

如果 CSV 打开乱码，可以尝试：

encoding="utf-8-sig"

本文存储层已经使用了这个编码。

如果网页正文乱码，requests 可以这样处理：

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

有些老网站可能使用 GBK，可以手动指定：

response.encoding = "gbk"

但不建议一开始就写死编码，最好先根据响应头和实际内容判断。

6. 为什么只抓到第一页？

可能原因有：

• 下一页选择器错误；
• 真实网站使用页码参数，但没有下一页按钮；
• 下一页链接是 JavaScript 动态生成；
• max_pages 设置太小；
• 目录页第二页没有展项。

本文代码有两种分页方式：

1. 优先使用页面里的下一页链接；
2. 如果没有下一页链接，就按 list_url_template.format(page=page) 生成页码。

如果真实网站分页参数不是 page，比如是 p 或 current，修改模板即可：

--list-url-template "https://www.example.com/exhibits?p={page}"

7. 字段都为空怎么办？

字段为空通常说明详情页选择器不对。

先保存详情页 HTML：

Path("detail_debug.html").write_text(detail_html, encoding="utf-8")

再检查目标字段真实结构。

比如页面可能不是：

<span class="field-age">8 岁以上</span>

而是：

<li>
  <label>适龄范围</label>
  <div class="value">8 岁以上</div>
</li>

对应选择器就要改成：

"适龄范围": "li.age .value"

如果字段没有单独 class，就需要更灵活的解析，例如按标签文本匹配。

---

1️⃣1️⃣ 进阶优化

1. 增加内容 hash 去重

URL 去重很实用，但不是万能的。如果网站 URL 改了，内容没变，就会重复入库。这时可以对核心字段做 hash。

import hashlib


def make_content_hash(item: dict) -> str:
    text = "|".join(
        [
            item.get("展项名", ""),
            item.get("学科主题", ""),
            item.get("适龄范围", ""),
            item.get("互动方式", ""),
        ]
    )
    return hashlib.md5(text.encode("utf-8")).hexdigest()

然后 CSV 增加一列：

内容指纹

如果使用数据库，可以对 hash 字段建立唯一索引。

2. 断点续跑

当前代码已经通过读取 CSV 中的“链接”实现了简单断点续跑。程序中断后再次运行，已保存的 URL 会跳过。

如果要更精细，可以维护两个文件：

data/seen_urls.txt
data/failed_urls.txt

失败链接单独记录，后续专门重试。

示例：

def append_failed_url(url: str, reason: str) -> None:
    with open("data/failed_urls.txt", "a", encoding="utf-8") as f:
        f.write(f"{url}\t{reason}\n")

3. 增加日志统计

现在日志已经记录保存数量和失败数量。进一步可以统计：

• 请求总数；
• 成功率；
• 平均响应时间；
• 每页展项数量；
• 字段缺失率。

字段缺失率对数据质量很重要。比如“适龄范围”缺失 60%，说明这个字段可能不是网站稳定字段，后续分析时要谨慎。

4. 使用 SQLite 存储

如果数据量超过几万条，CSV 的去重和更新会变得不方便。可以改用 SQLite。

建表示例：

CREATE TABLE IF NOT EXISTS exhibits (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    exhibit_name TEXT,
    subject_theme TEXT,
    age_range TEXT,
    interaction_mode TEXT,
    url TEXT UNIQUE,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP
);

Python 写入示例：

import sqlite3


class SQLiteStorage:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self.init_table()

    def init_table(self):
        sql = """
        CREATE TABLE IF NOT EXISTS exhibits (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            exhibit_name TEXT,
            subject_theme TEXT,
            age_range TEXT,
            interaction_mode TEXT,
            url TEXT UNIQUE,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        );
        """
        self.conn.execute(sql)
        self.conn.commit()

    def save_item(self, item: dict):
        sql = """
        INSERT OR IGNORE INTO exhibits (
            exhibit_name,
            subject_theme,
            age_range,
            interaction_mode,
            url
        ) VALUES (?, ?, ?, ?, ?);
        """
        self.conn.execute(
            sql,
            (
                item.get("展项名", ""),
                item.get("学科主题", ""),
                item.get("适龄范围", ""),
                item.get("互动方式", ""),
                item.get("链接", ""),
            ),
        )
        self.conn.commit()

5. 轻量并发

对这类内容站点，不建议高并发。但如果经过确认，访问频率允许，可以对详情页做低并发，比如 2–3 个线程。

示例思路：

from concurrent.futures import ThreadPoolExecutor, as_completed


def fetch_one_detail(fetcher, parser, detail_url):
    html = fetcher.fetch(detail_url)
    return parser.parse_detail(html, detail_url)


with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [
        executor.submit(fetch_one_detail, fetcher, parser, url)
        for url in detail_urls
    ]

    for future in as_completed(futures):
        item = future.result()
        storage.save_item(item)

注意，并发不是为了“冲得更快”，而是为了在网络等待时间较长时提高一点效率。并发数应该保守设置，并且配合限速。

6. 定时任务

如果你要每周更新一次展项目录，可以使用 cron。

Linux / macOS 示例：

crontab -e

添加：

0 9 * * 1 cd /path/to/science_exhibit_spider && .venv/bin/python run.py --demo >> logs/cron.log 2>&1

表示每周一上午 9 点运行一次。

生产环境里，也可以用 Airflow、Prefect 或其他任务调度系统。但对于一个小型目录采集项目，cron 已经够用。

7. 迁移到 Scrapy

当你需要以下能力时，可以考虑 Scrapy：

• 大量 URL 调度；
• 自动去重；
• 中间件；
• pipeline；
• 请求失败重试；
• 并发控制；
• 日志和统计；
• 更规范的项目结构。

Scrapy 版的核心思路仍然一样：

start_requests 生成目录页请求
parse 解析目录页和详情链接
parse_detail 解析详情字段
pipeline 写入 CSV / SQLite / MySQL

技术会变，但“目录页到详情页”的基本模型不会变。

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

本文完成了一个完整的科普馆展项目录分页爬虫。它从目录页开始，逐页解析展项详情链接，再进入详情页提取展项名、学科主题、适龄范围、互动方式和链接，最后导出为 CSV。

这套代码的重点不在于“抓了多少数据”，而在于建立了一种稳定的写法：

请求层负责稳定访问
解析层负责结构化抽取
存储层负责落盘和去重
调度层负责分页和节奏

这种分层思路很朴素，但非常耐用。真实项目里，页面会改版，字段会缺失，请求会失败，编码会出错。把这些情况提前纳入设计，爬虫才不会一遇到小波动就整段崩掉。

下一步可以继续做几件事：

1. 把 CSV 存储替换成 SQLite 或 MySQL；
2. 增加内容 hash，识别展项介绍是否更新；
3. 使用 Playwright 处理动态渲染页面；
4. 使用 Scrapy 重构项目，增强调度和并发能力；
5. 增加 pandas 分析脚本，统计学科主题和适龄范围分布；
6. 做一个简单的可视化看板，展示不同科普馆展项结构差异。

我自己的习惯是：先用 requests + BeautifulSoup 把链路跑通，再决定是否引入更重的框架。爬虫不是工具越复杂越好，而是要看目标站点、数据规模和维护成本。对于“科普馆展项目录分页”这种任务，本文这套实现已经足够作为一个扎实的起点。

最后提醒一句：技术可以提高效率，但边界感也很重要。采集公开信息时，保持低频、透明、克制，是一个爬虫开发者最基本的职业习惯。

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

---

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

---

✅ 免责声明

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

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

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