国家公园活动公告分页采集实战：用 Python 把活动名、时间、地点、主办方和报名说明结构化导出

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

全文目录：

• • 🌟 开篇语
  • 0️⃣ 前言（Preface）
  • 1️⃣ 摘要（Abstract）
  • 2️⃣ 背景与需求（Why）
  • 3️⃣ 合规与注意事项（必写）
  • • 3.1 robots.txt 基本说明
    • 3.2 频率控制，不要攻击式并发
    • 3.3 不采集敏感信息，不绕过付费或登录限制
  • 4️⃣ 技术选型与整体流程（What/How）
  • • 4.1 静态、动态、API，该怎么判断
    • 4.2 整体流程
    • 4.3 为什么选择 requests、BeautifulSoup 和 lxml
  • 5️⃣ 环境准备与依赖安装（可复现）
  • • 5.1 Python 版本
    • 5.2 安装依赖
    • 5.3 推荐项目结构
  • 6️⃣ 核心实现：请求层（Fetcher）
  • • 6.1 配置文件 config.py
    • 6.2 数据模型 models.py
    • 6.3 Fetcher 实现
  • 7️⃣ 核心实现：解析层（Parser）
  • • 7.1 Parser 代码
    • 7.2 列表页如何拿详情链接
    • 7.3 详情页如何抽字段
    • 7.4 缺失字段怎么办
  • 8️⃣ 数据存储与导出（Storage）
  • • 8.1 字段映射表
    • 8.2 去重策略
    • 8.3 storage.py
  • 9️⃣ 运行方式与结果展示（必写）
  • • 9.1 生成本地演示站点
    • 9.2 主入口 main.py
    • 9.3 启动命令
    • 9.4 输出位置
    • 9.5 示例结果
  • 🔟 常见问题与排错（强烈建议写）
  • • 10.1 403 怎么办
    • 10.2 429 怎么办
    • 10.3 HTML 抓到空壳怎么办
    • 10.4 解析报错怎么办
    • 10.5 编码或乱码如何处理
  • 1️⃣1️⃣ 进阶优化（可选但加分）
  • • 11.1 并发：线程、asyncio 或 Scrapy
    • 11.2 断点续跑
    • 11.3 日志与监控
    • 11.4 定时任务
    • 11.5 数据质量优化
  • 1️⃣2️⃣ 总结与延伸阅读
• 附录：完整文件清单
• • requirements.txt
  • config.py
  • models.py
  • fetcher.py
  • parser.py
  • storage.py
  • main.py
  • tools/make_demo_site.py
  • 🌟 文末
  • • ✅ 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集
    • ✅ 免责声明

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

这篇文章要做的事情很明确：使用 Python、requests、BeautifulSoup 和 lxml，采集一个“国家公园活动公告分页”类型的网站，把活动名、时间、地点、主办方、报名说明等字段整理成结构化 CSV 和 JSONL 文件。

读完本文，你可以获得三点：

第一，理解公告分页类页面的采集流程，不再只会“复制网页源码然后到处找标签”。

第二，拿到一套分层清晰、可直接运行的 Python 爬虫项目代码，包括请求层、解析层、清洗层和存储层。

第三，知道遇到 403、429、空壳 HTML、字段缺失、编码乱码、分页中断时应该怎么排查，而不是一报错就重写代码。

本文不讨论绕过限制、不讨论登录态破解，也不采集敏感信息。重点放在公共活动公告页面的信息整理和技术实现上。

1️⃣ 摘要（Abstract）

本文以“国家公园活动公告分页”为采集对象，使用 Python 的 requests、BeautifulSoup、lxml、csv、jsonlines 等工具，将活动公告中的活动名、时间、地点、主办方、报名说明提取出来，并导出为可分析的结构化数据文件。

读完本文，你会掌握：

这类项目看起来不复杂，但它非常适合作为爬虫基本功训练。因为公告页往往有分页、有详情页、有字段不整齐、有时间格式不统一，也经常出现“列表页有标题，详情页才有报名说明”的情况。把这个案例做稳，比盲目追求复杂框架更有价值。

2️⃣ 背景与需求（Why）

很多国家公园、自然保护地、湿地公园、森林公园、生态教育中心都会定期发布活动公告，例如自然观察、生态讲座、亲子科普、志愿者巡护、摄影分享、露营体验、巡展讲解等。对普通游客来说，逐页浏览公告并不麻烦；但如果你希望做信息聚合、活动日历、数据分析或自动提醒，手动整理就会很快变得低效。

我个人在处理这类网页时，最烦的不是“写爬虫”，而是网站每个公告页的格式略有不同：有的页面把时间写成“2026年6月15日 09:00-11:30”，有的写成“6月15日上午”，有的地点藏在正文第一段，有的报名说明是一个链接，还有的活动主办方只在正文末尾出现。看起来只是五个字段，真正写代码时会遇到不少边角问题。

本次实战目标是采集“国家公园活动公告分页”中的活动信息。我们假设目标站点具有如下结构：

• 列表页：按页展示活动公告卡片，每条卡片包含标题、发布时间或活动简述，并提供详情页链接。
• 详情页：展示活动完整内容，包含活动时间、地点、主办方和报名说明。
• 分页规则：类似 /events/page-1.html、/events/page-2.html，或者列表页中存在“下一页”链接。

目标字段如下：

为什么要做结构化？

一是方便分析。例如统计未来一个月有哪些活动、哪些园区活动密度高、哪些主办方发布频率高。

二是方便聚合。例如做一个统一的“国家公园活动日历”，把不同页面的公告转成统一格式。

三是方便自动化。例如每天早上定时抓取，发现新增活动后推送到邮件、飞书、企业微信或个人待办。

本篇文章使用的是“静态分页 + 详情页解析”的方案。真实站点如果是动态渲染，可以在此基础上切换到接口抓取或 Playwright 渲染，后面会专门说明。

3️⃣ 合规与注意事项（必写）

在写爬虫之前，先把边界说清楚。爬虫不是“能拿到就随便拿”，更不是用并发去压别人的网站。技术分享应该讲方法，也应该讲克制。

3.1 robots.txt 基本说明

robots.txt 是站点给网络爬虫的访问建议文件，通常位于网站根目录，例如：

https://example.com/robots.txt

里面可能出现类似内容：

User-agent: *
Disallow: /admin/
Disallow: /private/
Allow: /events/

这表示普通爬虫不应访问 /admin/ 和 /private/，但可以访问 /events/。实际项目中，建议在采集前先查看 robots.txt，确认目标路径是否允许访问。

本文代码会使用 Python 标准库 urllib.robotparser 做基本检查。它不是万能的，但足够作为一个良好的起点。

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

公告页通常不是高频更新页面，没有必要每秒请求几十次。合理做法是：

• 每个请求之间随机暂停 1 到 3 秒。
• 请求失败时指数退避，不要立即疯狂重试。
• 控制最大页数，避免分页异常导致无限抓取。
• 并发不是越高越好，公共信息页一般单线程或低并发就够。

我平时写这种采集脚本，会优先追求“稳”和“可恢复”，而不是追求“十秒抓完”。很多网站本身服务器配置不高，抓得太急既不礼貌，也容易触发限流。

3.3 不采集敏感信息，不绕过付费或登录限制

本案例只采集公开活动公告里的通用字段，例如活动名、时间、地点、主办方和报名说明。代码不涉及：

• 登录态绕过。
• 验证码绕过。
• 付费内容绕过。
• 个人隐私采集。
• 非公开接口探测。
• 高频压测式访问。

如果活动报名说明中包含联系电话或邮箱，是否存储要看具体场景。如果只是内部整理，可以保留；如果要公开展示，建议做脱敏处理，或者只展示“请到原页面查看报名方式”。

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

4.1 静态、动态、API，该怎么判断

爬网页之前，我一般先判断目标站点属于哪一类。

第一类是静态页面。浏览器打开页面后，活动列表和详情字段直接存在于 HTML 源码中。右键查看网页源代码，可以看到标题、时间、地点等文字。这类页面最适合用 requests + BeautifulSoup 或 lxml。

第二类是动态页面。网页初始 HTML 只是一个空壳，内容由 JavaScript 后续渲染出来。你在源代码里看不到活动标题，但在浏览器开发者工具的 Elements 面板里能看到。这类页面可以先找接口，找不到再考虑 Playwright。

第三类是接口页面。前端通过接口返回 JSON 数据，例如 /api/events?page=1。如果接口是公开的、无敏感限制，并且符合访问规范，直接请求 JSON 通常更稳定。

本文按第一类处理，也就是“静态分页列表 + 详情页”。原因很简单：国家公园活动公告类页面大多偏传统内容管理系统，静态 HTML 或服务端渲染比例较高。即使实际站点是动态页面，本文的分层结构也可以保留，只需要替换 Fetcher 或 Parser。

4.2 整体流程

本项目流程如下：

采集 Fetch
↓
解析 Parse
↓
清洗 Clean
↓
去重 Deduplicate
↓
存储 Store
↓
导出 CSV / JSONL

更细一点：

读取配置
↓
检查 robots.txt
↓
请求列表页
↓
解析详情页链接
↓
请求详情页
↓
抽取活动名、时间、地点、主办方、报名说明
↓
字段清洗与容错
↓
根据 URL 或内容 hash 去重
↓
写入 CSV 和 JSONL
↓
记录日志

4.3 为什么选择 requests、BeautifulSoup 和 lxml

这篇文章选用的技术栈是：

• requests：负责 HTTP 请求，API 简单，适合静态页面。
• BeautifulSoup：负责 HTML 解析，写法直观，容错能力较好。
• lxml：作为底层解析器，速度比 Python 内置 html.parser 更好。
• csv：Python 标准库，适合快速导出表格。
• jsonlines：每行一个 JSON，便于追加写入和流式处理。
• tenacity：可选，用于更灵活的重试。本项目使用 requests 自带 Retry 思路，也会给出退避写法。

为什么不用 Scrapy？

Scrapy 很强，适合大规模、复杂站点和长期任务。但本文目标是让读者能清楚理解请求层、解析层、存储层的基本逻辑。先用 requests 写完整，比直接套 Scrapy 模板更利于打基础。

为什么不用 Playwright？

Playwright 适合动态渲染页面，但它启动浏览器，资源开销比 requests 大。如果目标页面源码里已经有数据，就没必要用浏览器渲染。我的习惯是：先 requests，后接口，最后才 Playwright。

5️⃣ 环境准备与依赖安装（可复现）

5.1 Python 版本

推荐使用 Python 3.10 或以上版本。本文代码在 Python 3.11 思路下编写，Python 3.10 也可以正常运行。

查看版本：

python –version

建议创建虚拟环境：

python -m venv .venv

Windows 激活：

.venv\\Scripts\\activate

macOS / Linux 激活：

source .venv/bin/activate

5.2 安装依赖

创建 requirements.txt：

requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.3.0
jsonlines==4.0.0
python-dateutil==2.9.0.post0

安装：

pip install -r requirements.txt

5.3 推荐项目结构

项目目录建议如下：

park_events_spider/
├── README.md
├── requirements.txt
├── config.py
├── fetcher.py
├── parser.py
├── storage.py
├── models.py
├── main.py
├── tools/
│ └── make_demo_site.py
├── output/
│ ├── park_events.csv
│ └── park_events.jsonl
└── logs/
└── spider.log

为了让代码真实可运行，本文还准备了一个本地演示站点生成脚本 tools/make_demo_site.py。你可以先生成本地 HTML 页面，用 Python 自带的 HTTP Server 启动，再运行爬虫。等逻辑跑通后，把 BASE_URL 和选择器替换成真实站点即可。

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

请求层主要解决四件事：

6.1 配置文件 config.py

# config.py
from pathlib import Path
import os

BASE_DIR = Path(__file__).resolve().parent

# 本地 demo 默认地址。真实采集时可以通过环境变量替换。
# 例如：
# export PARK_EVENTS_BASE_URL="https://example.com/events/page-{page}.html"
BASE_URL = os.getenv(
"PARK_EVENTS_BASE_URL",
"http://127.0.0.1:8000/events/page-{page}.html"
)

# 最大采集页数，避免分页异常时无限抓取
MAX_PAGES = int(os.getenv("PARK_EVENTS_MAX_PAGES", "5"))

# 请求间隔，单位秒
MIN_DELAY = float(os.getenv("PARK_EVENTS_MIN_DELAY", "1.0"))
MAX_DELAY = float(os.getenv("PARK_EVENTS_MAX_DELAY", "2.5"))

# 请求超时
CONNECT_TIMEOUT = 5
READ_TIMEOUT = 20

# 输出目录
OUTPUT_DIR = BASE_DIR / "output"
OUTPUT_DIR.mkdir(exist_ok=True)

CSV_PATH = OUTPUT_DIR / "park_events.csv"
JSONL_PATH = OUTPUT_DIR / "park_events.jsonl"

# 日志目录
LOG_DIR = BASE_DIR / "logs"
LOG_DIR.mkdir(exist_ok=True)
LOG_PATH = LOG_DIR / "spider.log"

# 选择器配置：真实站点只需要重点调整这里
LIST_ITEM_SELECTOR = ".event-card"
DETAIL_LINK_SELECTOR = "a.event-title"
NEXT_PAGE_SELECTOR = "a.next"

DETAIL_TITLE_SELECTOR = "h1"
DETAIL_CONTENT_SELECTOR = ".article-content"

# 字段标签。详情页里常见写法包括“活动时间：”“时间：”“活动地点：”等。
FIELD_LABELS = {
"time": ["活动时间", "时间", "举办时间"],
"location": ["活动地点", "地点", "集合地点"],
"organizer": ["主办方", "主办单位", "组织单位"],
"registration": ["报名说明", "报名方式", "如何报名", "报名须知"],
}

这份配置有两个好处。

第一，站点变化时不用到处改代码。选择器集中放在一个文件里，维护成本更低。

第二，适合多人协作。一个人负责调整选择器，一个人负责请求层，另一个人负责数据分析，不会互相踩代码。

6.2 数据模型 models.py

# models.py
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional
import hashlib

@dataclass
class EventItem:
name: str
time: Optional[str]
location: Optional[str]
organizer: Optional[str]
registration: Optional[str]
detail_url: str
summary: Optional[str]
crawled_at: str

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

def content_hash(self) –> str:
raw = "|".join([
self.name or "",
self.time or "",
self.location or "",
self.organizer or "",
self.registration or "",
self.detail_url or "",
])
return hashlib.sha256(raw.encode("utf-8")).hexdigest()

数据模型不要写得太复杂。爬虫项目早期最容易犯的错误，是一上来就设计一堆继承和抽象类。我的建议是先用 dataclass，把字段固定下来，跑通之后再考虑扩展。

6.3 Fetcher 实现

# fetcher.py
import logging
import random
import time
from urllib.parse import urlparse
from urllib.robotparser import RobotFileParser

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from config import (
CONNECT_TIMEOUT,
READ_TIMEOUT,
MIN_DELAY,
MAX_DELAY,
)

class Fetcher:
def __init__(self, user_agent: str = None):
self.logger = logging.getLogger(self.__class__.__name__)
self.session = requests.Session()

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

self.session.headers.update({
"User-Agent": self.user_agent,
"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",
"Connection": "keep-alive",
})

retry = Retry(
total=3,
connect=3,
read=3,
status=3,
backoff_factor=1.2,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET"],
raise_on_status=False,
)

adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)

self.robot_cache = {}

def sleep_polite(self):
seconds = random.uniform(MIN_DELAY, MAX_DELAY)
self.logger.debug("Sleep %.2f seconds before next request", seconds)
time.sleep(seconds)

def _robots_url(self, url: str) –> str:
parsed = urlparse(url)
return f"{parsed.scheme}://{parsed.netloc}/robots.txt"

def allowed_by_robots(self, url: str) –> bool:
"""
基础 robots.txt 检查。
本地 demo 或 robots 不可访问时，不直接判定为禁止，而是记录日志并继续。
实际生产中可以根据团队规范改成更严格模式。
"""
parsed = urlparse(url)

if parsed.hostname in {"127.0.0.1", "localhost"}:
return True

robots_url = self._robots_url(url)

if robots_url not in self.robot_cache:
rp = RobotFileParser()
rp.set_url(robots_url)
try:
rp.read()
self.robot_cache[robots_url] = rp
except Exception as exc:
self.logger.warning("Read robots.txt failed: %s, error=%s", robots_url, exc)
return True

rp = self.robot_cache[robots_url]
return rp.can_fetch(self.user_agent, url)

def get(self, url: str, referer: str = None) –> str:
if not self.allowed_by_robots(url):
raise PermissionError(f"Blocked by robots.txt: {url}")

headers = {}
if referer:
headers["Referer"] = referer

self.sleep_polite()

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

response = self.session.get(
url,
headers=headers,
timeout=(CONNECT_TIMEOUT, READ_TIMEOUT),
)

if response.status_code == 403:
self.logger.warning("403 Forbidden: %s", url)
elif response.status_code == 429:
self.logger.warning("429 Too Many Requests: %s", url)

response.raise_for_status()

# requests 会根据响应头猜编码；如果猜错，可以手动设置。
if not response.encoding or response.encoding.lower() == "iso-8859-1":
response.encoding = response.apparent_encoding

return response.text

这里的几个细节值得单独说一下。

headers 不是为了“伪装攻击”，而是为了让请求更接近正常浏览器访问。有些站点会拒绝没有 User-Agent 的请求，这种拒绝并不罕见。

timeout 必须写。没有 timeout 的爬虫在网络抖动时会卡很久，定时任务尤其容易出问题。

Session 有两个意义。一是复用 TCP 连接，效率更好。二是如果站点正常使用 cookie 维护访问状态，Session 可以自然保存 cookie。本文不涉及登录态，但 Session 依旧是好习惯。

重试要克制。失败了重试三次可以理解，失败后每秒重试几百次就不合适。对 429 这样的限流状态，更应该降低频率，而不是提高频率。

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

解析层负责把 HTML 变成字段。它最考验耐心，因为真实页面往往不如样例整齐。

本文使用 BeautifulSoup + CSS Selector。原因是公告页标签结构通常比较简单，用 CSS Selector 可读性高，也方便后续维护。如果你更熟悉 XPath，也可以换成 lxml 的 XPath。

7.1 Parser 代码

# parser.py
import re
from datetime import datetime
from typing import List, Optional
from urllib.parse import urljoin

from bs4 import BeautifulSoup

from config import (
LIST_ITEM_SELECTOR,
DETAIL_LINK_SELECTOR,
NEXT_PAGE_SELECTOR,
DETAIL_TITLE_SELECTOR,
DETAIL_CONTENT_SELECTOR,
FIELD_LABELS,
)
from models import EventItem

def clean_text(text: Optional[str]) –> Optional[str]:
if text is None:
return None
text = re.sub(r"\\s+", " ", text)
text = text.replace("\\u3000", " ")
text = text.strip(" ：:;；\\n\\t")
return text or None

def compact_text(text: Optional[str], limit: int = 140) –> Optional[str]:
text = clean_text(text)
if not text:
return None
return text[:limit] + ("…" if len(text) > limit else "")

def soupify(html: str) –> BeautifulSoup:
return BeautifulSoup(html, "lxml")

class EventParser:
def parse_list_page(self, html: str, base_url: str) –> List[str]:
"""
从列表页提取详情页链接。
"""
soup = soupify(html)
links = []

cards = soup.select(LIST_ITEM_SELECTOR)

# 如果卡片选择器失效，做一个兜底：直接找包含 href 的活动标题链接。
if not cards:
for a in soup.select(DETAIL_LINK_SELECTOR):
href = a.get("href")
if href:
links.append(urljoin(base_url, href))
return list(dict.fromkeys(links))

for card in cards:
a = card.select_one(DETAIL_LINK_SELECTOR)
if not a:
continue

href = a.get("href")
if not href:
continue

detail_url = urljoin(base_url, href)
links.append(detail_url)

# 保序去重
return list(dict.fromkeys(links))

def parse_next_page(self, html: str, base_url: str) –> Optional[str]:
"""
如果站点提供“下一页”链接，可以从 HTML 中解析出来。
本文主流程使用 page-{page}.html 模板，但这个方法保留给真实站点。
"""
soup = soupify(html)
a = soup.select_one(NEXT_PAGE_SELECTOR)
if a and a.get("href"):
return urljoin(base_url, a["href"])
return None

def parse_detail_page(self, html: str, detail_url: str) –> EventItem:
soup = soupify(html)

title = self._extract_title(soup)
content_node = soup.select_one(DETAIL_CONTENT_SELECTOR)

if content_node:
content_text = content_node.get_text("\\n", strip=True)
else:
# 兜底：有些站点正文没有明显容器，直接取 body。
body = soup.select_one("body")
content_text = body.get_text("\\n", strip=True) if body else soup.get_text("\\n", strip=True)

content_text = clean_text(content_text) or ""

field_map = self._extract_fields_from_content(content_text)

item = EventItem(
name=title or self._guess_title_from_content(content_text) or "未命名活动",
time=field_map.get("time"),
location=field_map.get("location"),
organizer=field_map.get("organizer"),
registration=field_map.get("registration"),
detail_url=detail_url,
summary=compact_text(content_text, limit=180),
crawled_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
)

return item

def _extract_title(self, soup: BeautifulSoup) –> Optional[str]:
title_node = soup.select_one(DETAIL_TITLE_SELECTOR)
if title_node:
return clean_text(title_node.get_text(" ", strip=True))

if soup.title:
return clean_text(soup.title.get_text(" ", strip=True))

return None

def _guess_title_from_content(self, content: str) –> Optional[str]:
if not content:
return None

lines = [clean_text(x) for x in content.split("\\n")]
lines = [x for x in lines if x]

if not lines:
return None

# 通常第一行是标题或导语。这里取第一行作为兜底标题。
return lines[0][:80]

def _extract_fields_from_content(self, content: str) –> dict:
"""
从正文文本中抽取字段。
支持类似：
活动时间：2026年6月15日 09:00-11:30
活动地点：青溪国家公园访客中心
主办方：青溪国家公园管理处
报名说明：请通过官网预约，名额30人，额满即止。
"""
result = {
"time": None,
"location": None,
"organizer": None,
"registration": None,
}

if not content:
return result

# 把常见中文冒号、英文冒号统一考虑。
lines = re.split(r"[\\n\\r]+|(?<=。)", content)
lines = [clean_text(line) for line in lines]
lines = [line for line in lines if line]

for key, labels in FIELD_LABELS.items():
value = self._find_value_by_labels(lines, labels)
if value:
result[key] = value

# 如果报名说明没有通过标签匹配到，尝试从包含“报名”的句子中兜底。
if not result["registration"]:
result["registration"] = self._guess_registration(content)

return result

def _find_value_by_labels(self, lines: List[str], labels: List[str]) –> Optional[str]:
for line in lines:
for label in labels:
pattern = rf"^{re.escape(label)}\\s*[:：]\\s*(.+)$"
m = re.search(pattern, line)
if m:
return clean_text(m.group(1))

# 兼容字段不在行首的情况：
# 例如：集合信息：活动地点：访客中心一楼
pattern2 = rf"{re.escape(label)}\\s*[:：]\\s*([^。；;]+)"
m2 = re.search(pattern2, line)
if m2:
return clean_text(m2.group(1))

return None

def _guess_registration(self, content: str) –> Optional[str]:
sentences = re.split(r"[。！？!?]\\s*", content)
candidates = []
for sentence in sentences:
if any(word in sentence for word in ["报名", "预约", "名额", "参加方式", "二维码"]):
candidates.append(clean_text(sentence))

if not candidates:
return None

# 取前两句，避免报名说明太长。
return "；".join([x for x in candidates if x][:2])

7.2 列表页如何拿详情链接

核心逻辑是：

cards = soup.select(".event-card")
for card in cards:
a = card.select_one("a.event-title")
href = a.get("href")

真实页面里，列表项可能叫 .news-item、.list-item、.article-list li，详情链接可能是 .title a 或 h3 a。所以本文把选择器放在 config.py 里。换站时优先改配置，不要改主流程。

7.3 详情页如何抽字段

详情页抽字段有两种常见方式。

第一种是结构化 HTML，例如：

<ul class="event-info">
<li><span>活动时间：</span>2026年6月15日</li>
<li><span>活动地点：</span>访客中心</li>
</ul>

这种最舒服，可以直接用 CSS Selector 找对应标签。

第二种是正文自然语言，例如：

活动时间：2026年6月15日 09:00-11:30
活动地点：青溪国家公园访客中心
主办方：青溪国家公园管理处
报名说明：请通过官网预约，名额30人。

本文按第二种写，因为它更常见，也更考验容错。解析思路是：先把正文文本按换行和句号拆开，然后用字段标签做正则匹配。

7.4 缺失字段怎么办

字段缺失是正常情况，不要因为某一个字段缺失就让整个详情页报错。本文的处理方式是：

• 活动名缺失：用正文第一行兜底，再不行填“未命名活动”。
• 时间缺失：填 None。
• 地点缺失：填 None。
• 主办方缺失：填 None。
• 报名说明缺失：尝试从包含“报名、预约、名额、二维码”的句子里推断。
• 详情页 URL：必须保留，因为它是溯源依据。

这种策略看起来朴素，但实战里很好用。数据采集不是追求每一条都完美，而是先保证整体数据可用，再针对异常样本做二次规则优化。

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

本文同时导出 CSV 和 JSONL。

CSV 适合 Excel、WPS、数据分析工具直接打开。

JSONL 适合程序继续处理，每一行都是一个 JSON 对象，追加写入和流式读取都方便。

8.1 字段映射表

8.2 去重策略

本文采用两层去重：

第一层，URL 去重。同一个详情页只处理一次。

第二层，内容 hash 去重。如果两个 URL 不同但字段内容完全一致，可以根据 hash 判断重复。

实际项目中，URL 去重最常用。如果站点改版导致同一活动有多个 URL，再增加内容 hash 更稳。

8.3 storage.py

# storage.py
import csv
import jsonlines
from pathlib import Path
from typing import Iterable, List, Set

from models import EventItem

FIELDNAMES = [
"name",
"time",
"location",
"organizer",
"registration",
"detail_url",
"summary",
"crawled_at",
]

class EventStorage:
def __init__(self, csv_path: Path, jsonl_path: Path):
self.csv_path = csv_path
self.jsonl_path = jsonl_path

self.csv_path.parent.mkdir(parents=True, exist_ok=True)
self.jsonl_path.parent.mkdir(parents=True, exist_ok=True)

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

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

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

return seen

def save_csv(self, items: Iterable[EventItem], append: bool = True) –> int:
items = list(items)
if not items:
return 0

file_exists = self.csv_path.exists()
mode = "a" if append else "w"

with self.csv_path.open(mode, encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=FIELDNAMES)

if not file_exists or not append:
writer.writeheader()

for item in items:
writer.writerow(item.to_dict())

return len(items)

def save_jsonl(self, items: Iterable[EventItem], append: bool = True) –> int:
items = list(items)
if not items:
return 0

mode = "a" if append else "w"

with jsonlines.open(self.jsonl_path, mode=mode) as writer:
for item in items:
writer.write(item.to_dict())

return len(items)

def save_all(self, items: List[EventItem]) –> int:
count_csv = self.save_csv(items, append=True)
self.save_jsonl(items, append=True)
return count_csv

这里 CSV 使用 utf-8-sig 编码，是为了让 Excel 打开时不容易乱码。严格来说，utf-8 已经足够，但很多办公环境还是会遇到 Excel 识别编码不准的问题，utf-8-sig 更省心。

9️⃣ 运行方式与结果展示（必写）

为了让这套代码可以直接运行，我们先做一个本地 demo 站点。这个 demo 会生成分页列表和详情页，结构模拟国家公园活动公告页面。

9.1 生成本地演示站点

创建 tools/make_demo_site.py：

# tools/make_demo_site.py
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parents[1]
SITE_DIR = BASE_DIR / "demo_site"
EVENTS_DIR = SITE_DIR / "events"

DETAIL_TEMPLATE = """<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>{title}</title>
</head>
<body>
<main class="page">
<article class="article-content">
<h1>{title}</h1>
<p>活动时间：{time}</p>
<p>活动地点：{location}</p>
<p>主办方：{organizer}</p>
<p>报名说明：{registration}</p>
<p>{intro}</p>
<p>请参与者根据天气情况准备舒适鞋服，并遵守现场工作人员指引。</p>
</article>
</main>
</body>
</html>
"""

LIST_TEMPLATE = """<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>国家公园活动公告 – 第 {page} 页</title>
</head>
<body>
<section class="event-list">
<h1>国家公园活动公告</h1>
{cards}
</section>
<nav class="pagination">
{next_link}
</nav>
</body>
</html>
"""

CARD_TEMPLATE = """
<div class="event-card">
<h2><a class="event-title" href="{href}">{title}</a></h2>
<p class="event-date">{time}</p>
<p class="event-desc">{desc}</p>
</div>
"""

EVENTS = [
{
"title": "夜观昆虫：国家公园夏季自然观察活动",
"time": "2026年6月15日 19:30-21:00",
"location": "青溪国家公园访客中心集合",
"organizer": "青溪国家公园管理处",
"registration": "请通过官网活动预约入口报名，名额30人，额满即止。",
"intro": "本次活动面向自然观察爱好者，带领参与者认识夜间昆虫和栖息地环境。",
},
{
"title": "高山植物识别公益讲解",
"time": "2026年6月18日 09:00-11:30",
"location": "云岭国家公园科普步道",
"organizer": "云岭国家公园自然教育中心",
"registration": "无需缴费，需提前一天电话预约，团队人数不超过20人。",
"intro": "讲解员将沿科普步道介绍常见高山植物的形态特征和生态价值。",
},
{
"title": "湿地候鸟监测志愿者体验日",
"time": "2026年6月22日 07:30-10:30",
"location": "湖湾湿地观鸟平台",
"organizer": "湖湾国家公园志愿服务队",
"registration": "报名时请填写姓名、联系方式和是否自备望远镜。",
"intro": "活动将介绍基础鸟类识别方法，并体验标准化监测记录流程。",
},
{
"title": "亲子自然笔记工作坊",
"time": "2026年6月28日 14:00-16:30",
"location": "松谷国家公园自然教室",
"organizer": "松谷国家公园宣教科",
"registration": "适合8岁以上儿童家庭参加，每组家庭最多2名成人陪同。",
"intro": "老师会带领孩子观察叶片、果实和昆虫痕迹，并完成一页自然笔记。",
},
{
"title": "国家公园巡护员的一天主题分享",
"time": "2026年7月3日 10:00-11:30",
"location": "青溪国家公园报告厅",
"organizer": "青溪国家公园管理处",
"registration": "现场座位有限，建议提前在线预约。",
"intro": "一线巡护员将分享巡护路线、野外记录和公众参与保护的方式。",
},
{
"title": "溪流生态观察与水质小实验",
"time": "2026年7月6日 09:30-12:00",
"location": "石桥溪生态观察点",
"organizer": "青溪国家公园自然教育团队",
"registration": "活动需穿防滑鞋，不建议6岁以下儿童参加。",
"intro": "活动通过简易水质测试和底栖动物观察，认识溪流生态系统。",
},
]

def build():
EVENTS_DIR.mkdir(parents=True, exist_ok=True)

for idx, event in enumerate(EVENTS, start=1):
detail_html = DETAIL_TEMPLATE.format(**event)
detail_path = EVENTS_DIR / f"detail-{idx}.html"
detail_path.write_text(detail_html, encoding="utf-8")

page_size = 3
pages = [EVENTS[i:i + page_size] for i in range(0, len(EVENTS), page_size)]

for page_index, page_events in enumerate(pages, start=1):
cards = []
for event in page_events:
idx = EVENTS.index(event) + 1
card = CARD_TEMPLATE.format(
href=f"detail-{idx}.html",
title=event["title"],
time=event["time"],
desc=event["intro"],
)
cards.append(card)

next_link = ""
if page_index < len(pages):
next_link = f'<a class="next" href="page-{page_index + 1}.html">下一页</a>'

list_html = LIST_TEMPLATE.format(
page=page_index,
cards="\\n".join(cards),
next_link=next_link,
)

page_path = EVENTS_DIR / f"page-{page_index}.html"
page_path.write_text(list_html, encoding="utf-8")

print(f"Demo site generated at: {SITE_DIR}")

if __name__ == "__main__":
build()

生成 demo 页面：

python tools/make_demo_site.py

启动本地服务：

python -m http.server 8000 -d demo_site

浏览器访问：

http://127.0.0.1:8000/events/page-1.html

能看到页面后，再打开另一个终端运行爬虫。

9.2 主入口 main.py

# main.py
import logging
from urllib.parse import urljoin

from config import (
BASE_URL,
MAX_PAGES,
CSV_PATH,
JSONL_PATH,
LOG_PATH,
)
from fetcher import Fetcher
from parser import EventParser
from storage import EventStorage

def setup_logging():
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
handlers=[
logging.FileHandler(LOG_PATH, encoding="utf-8"),
logging.StreamHandler(),
],
)

def build_list_url(page: int) –> str:
if "{page}" in BASE_URL:
return BASE_URL.format(page=page)
if page == 1:
return BASE_URL
separator = "&" if "?" in BASE_URL else "?"
return f"{BASE_URL}{separator}page={page}"

def run():
setup_logging()

logger = logging.getLogger("park_events_spider")

fetcher = Fetcher()
parser = EventParser()
storage = EventStorage(CSV_PATH, JSONL_PATH)

seen_urls = storage.load_seen_urls()
logger.info("Loaded seen urls: %d", len(seen_urls))

all_new_items = []
current_url = None

for page in range(1, MAX_PAGES + 1):
list_url = current_url or build_list_url(page)

try:
html = fetcher.get(list_url)
except Exception as exc:
logger.exception("Fetch list page failed: page=%s url=%s error=%s", page, list_url, exc)
break

detail_urls = parser.parse_list_page(html, list_url)

if not detail_urls:
logger.info("No detail urls found, stop. page=%s url=%s", page, list_url)
break

logger.info("Page %s found %d detail urls", page, len(detail_urls))

page_items = []

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.get(detail_url, referer=list_url)
item = parser.parse_detail_page(detail_html, detail_url)
except Exception as exc:
logger.exception("Fetch or parse detail failed: url=%s error=%s", detail_url, exc)
continue

seen_urls.add(detail_url)
page_items.append(item)

logger.info(
"Parsed event: name=%s time=%s location=%s",
item.name,
item.time,
item.location,
)

if page_items:
saved_count = storage.save_all(page_items)
all_new_items.extend(page_items)
logger.info("Saved %d items from page %s", saved_count, page)

# 如果真实站点是 next 链接翻页，可以启用这个逻辑。
next_url = parser.parse_next_page(html, list_url)

# 当前配置以 page 模板为主。为了 demo 简洁，这里不强制依赖 next_url。
# 如果 BASE_URL 没有 {page}，则使用 next_url 继续翻页。
if "{page}" not in BASE_URL:
if not next_url:
logger.info("No next page found, stop.")
break
current_url = next_url

logger.info("Spider finished. New items: %d", len(all_new_items))
logger.info("CSV output: %s", CSV_PATH)
logger.info("JSONL output: %s", JSONL_PATH)

if __name__ == "__main__":
run()

9.3 启动命令

完整运行顺序如下：

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

python tools/make_demo_site.py
python -m http.server 8000 -d demo_site

打开另一个终端：

source .venv/bin/activate
python main.py

Windows 用户把 source .venv/bin/activate 换成：

.venv\\Scripts\\activate

9.4 输出位置

运行后会生成：

output/park_events.csv
output/park_events.jsonl
logs/spider.log

9.5 示例结果

CSV 中可能看到类似结果：

JSONL 示例：

{"name":"夜观昆虫：国家公园夏季自然观察活动","time":"2026年6月15日 19:30-21:00","location":"青溪国家公园访客中心集合","organizer":"青溪国家公园管理处","registration":"请通过官网活动预约入口报名，名额30人，额满即止。","detail_url":"http://127.0.0.1:8000/events/detail-1.html","summary":"夜观昆虫：国家公园夏季自然观察活动 活动时间：2026年6月15日 19:30-21:00 活动地点：青溪国家公园访客中心集合…","crawled_at":"2026-06-08 10:30:12"}

🔟 常见问题与排错（强烈建议写）

10.1 403 怎么办

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

• 没有 User-Agent。
• Referer 不符合站点预期。
• 请求频率太高。
• 目标路径不希望被自动化访问。
• IP 触发了风控。

排查建议：

第一，先用浏览器正常打开目标页面，确认页面公开可访问。

第二，检查 robots.txt 和网站使用说明，确认目标路径适合采集。

第三，给请求加上合理 headers，尤其是 User-Agent、Accept-Language、Referer。

第四，降低请求频率，不要连续快速访问。

第五，如果仍然 403，不要继续强行尝试。很多时候，这表示站点并不欢迎自动化访问。可以改成手动下载公开数据、联系站点获取开放数据，或者只做本地样例练习。

10.2 429 怎么办

429 是 Too Many Requests，也就是请求太多。它和 403 不同，429 通常明确表示你访问太快了。

处理方式：

• 增大 MIN_DELAY 和 MAX_DELAY。
• 减少并发。
• 失败后指数退避，例如等待 2 秒、4 秒、8 秒。
• 如果响应头里有 Retry-After，优先按它提示等待。
• 不要用密集代理池去绕限流。本文定位是公共信息采集和技术学习，不做这种处理。

可以在 Fetcher 中增加一个简单逻辑：

if response.status_code == 429:
retry_after = response.headers.get("Retry-After")
if retry_after and retry_after.isdigit():
time.sleep(int(retry_after))

不过通常 requests 的 Retry 已经可以覆盖一部分情况。真正关键的还是控制频率。

10.3 HTML 抓到空壳怎么办

有时候你用 requests 抓到的 HTML 只有：

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

这说明页面内容可能由 JavaScript 动态渲染。处理顺序建议是：

第一，打开浏览器开发者工具，进入 Network 面板。

第二，刷新页面，筛选 Fetch / XHR 请求。

第三，看是否有返回 JSON 的接口，例如：

/api/events?page=1
/api/notice/list?pageNum=1&pageSize=10

第四，如果接口公开、无敏感限制、无需登录，可以优先请求接口。

第五，如果找不到接口，再考虑 Playwright 渲染页面。

Playwright 的思路如下：

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()
page.goto(url, wait_until="networkidle", timeout=30000)
html = page.content()
browser.close()
return html

但不要一遇到页面就上 Playwright。它更重，速度更慢，也更容易被环境问题影响。能用接口就用接口，能用 requests 就用 requests。

10.4 解析报错怎么办

解析报错常见原因有三个。

第一，选择器变了。比如原来标题是 h1，后来变成 .article-title。

第二，页面结构不统一。比如有些详情页是活动公告，有些详情页是普通新闻。

第三，字段标签变化。比如“主办方”写成了“主办单位”或“组织单位”。

解决方式：

• 把选择器集中放在配置文件。
• 对字段抽取做多标签匹配。
• 缺失字段填 None，不要中断整个任务。
• 把失败 URL 写入日志，后续人工抽样检查。
• 为典型页面保存 HTML 样本，方便回归测试。

解析层可以增加调试输出：

if not item.time:
logger.warning("Missing time field: %s", detail_url)

不要害怕脏数据。爬虫工程的日常就是和结构变化打交道。关键是日志要清楚，错误要可定位。

10.5 编码或乱码如何处理

如果页面出现乱码，先看响应头：

print(response.headers.get("Content-Type"))
print(response.encoding)
print(response.apparent_encoding)

可以手动设置：

response.encoding = response.apparent_encoding

如果你明确知道站点是 UTF-8：

response.encoding = "utf-8"

CSV 打开乱码时，使用 utf-8-sig：

with open("output.csv", "w", encoding="utf-8-sig", newline="") as f:
...

有些旧站点可能是 GBK：

response.encoding = "gb18030"

不要盲目到处 .encode().decode()，那样很容易把数据越修越乱。先判断源编码，再决定怎么处理。

1️⃣1️⃣ 进阶优化（可选但加分）

基础版能跑以后，可以继续优化。这里给几个实用方向。

11.1 并发：线程、asyncio 或 Scrapy

公告页采集通常不需要高并发，但如果目标站点页面很多，可以做低并发优化。

线程池版本示例：

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_one_detail(fetcher, parser, detail_url, referer):
html = fetcher.get(detail_url, referer=referer)
return parser.parse_detail_page(html, detail_url)

def fetch_details_concurrently(fetcher, parser, detail_urls, referer, max_workers=3):
items = []

with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_map = {
executor.submit(fetch_one_detail, fetcher, parser, url, referer): url
for url in detail_urls
}

for future in as_completed(future_map):
url = future_map[future]
try:
item = future.result()
items.append(item)
except Exception as exc:
print(f"Detail failed: {url}, error={exc}")

return items

这里的 max_workers=3 是比较保守的设置。不要一上来开 50、100。对于公共公告站点，低并发已经足够。

如果项目升级到 Scrapy，可以获得：

• 内置调度器。
• 自动去重。
• 中间件。
• 下载延迟。
• 管道存储。
• 更适合长期运行的结构。

Scrapy 的 Item 可以这样设计：

import scrapy

class ParkEventItem(scrapy.Item):
name = scrapy.Field()
time = scrapy.Field()
location = scrapy.Field()
organizer = scrapy.Field()
registration = scrapy.Field()
detail_url = scrapy.Field()
summary = scrapy.Field()
crawled_at = scrapy.Field()

不过对于本文这个规模，requests 版本更容易讲清楚，也更方便新人调试。

11.2 断点续跑

断点续跑的核心是保存“已经抓过什么”。

本文已经通过 CSV 里的 detail_url 做了基本去重。更稳一点，可以单独维护一个 seen 文件：

from pathlib import Path

SEEN_PATH = Path("output/seen_urls.txt")

def load_seen():
if not SEEN_PATH.exists():
return set()
return set(x.strip() for x in SEEN_PATH.read_text(encoding="utf-8").splitlines() if x.strip())

def append_seen(url):
with SEEN_PATH.open("a", encoding="utf-8") as f:
f.write(url + "\\n")

如果你使用 SQLite，可以建一张表：

CREATE TABLE IF NOT EXISTS park_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
time TEXT,
location TEXT,
organizer TEXT,
registration TEXT,
detail_url TEXT UNIQUE NOT NULL,
summary TEXT,
crawled_at TEXT
);

SQLite 版本存储示例：

import sqlite3
from pathlib import Path

DB_PATH = Path("output/park_events.db")

def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS park_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
time TEXT,
location TEXT,
organizer TEXT,
registration TEXT,
detail_url TEXT UNIQUE NOT NULL,
summary TEXT,
crawled_at TEXT
)
""")
conn.commit()
return conn

def insert_item(conn, item):
sql = """
INSERT OR IGNORE INTO park_events
(name, time, location, organizer, registration, detail_url, summary, crawled_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
"""
conn.execute(sql, (
item.name,
item.time,
item.location,
item.organizer,
item.registration,
item.detail_url,
item.summary,
item.crawled_at,
))
conn.commit()

SQLite 的好处是去重方便，查询也方便。例如查未来活动、按主办方统计，都比直接扫 CSV 更稳。

11.3 日志与监控

爬虫跑起来之后，最怕“看起来跑完了，其实一条没抓到”。所以日志至少要记录：

• 请求了哪个列表页。
• 每页发现多少详情链接。
• 每个详情页解析是否成功。
• 缺失字段数量。
• 保存了多少条。
• 失败 URL 和异常信息。

可以增加统计信息：

stats = {
"list_pages": 0,
"detail_pages": 0,
"saved_items": 0,
"failed_items": 0,
"missing_time": 0,
"missing_location": 0,
}

每次结束时打印：

logger.info("Stats: %s", stats)

如果要做长期任务，可以把日志接入：

• 文件日志轮转。
• Prometheus。
• Grafana。
• 企业微信机器人。
• 邮件通知。

小项目不一定需要完整监控，但至少要有可读日志。

11.4 定时任务

Linux 上可以用 cron：

crontab -e

每天早上 8 点运行：

0 8 * * * cd /path/to/park_events_spider && /path/to/.venv/bin/python main.py >> logs/cron.log 2>&1

如果在 Windows 上，可以使用任务计划程序。

如果任务之间有依赖，例如“先抓取、再清洗、再发邮件、再更新看板”，可以考虑 Airflow、Prefect 或 Dagster。但不要为了一个小爬虫过早引入复杂调度系统。技术选型要服务问题，不是服务排场。

11.5 数据质量优化

活动公告这类数据很容易出现时间格式不统一。可以加一个时间标准化函数：

import re
from dateutil import parser as date_parser

def normalize_event_time(time_text: str):
if not time_text:
return None

text = time_text.strip()

# 示例：2026年6月15日 19:30-21:00
text = text.replace("年", "-").replace("月", "-").replace("日", " ")

# 这里只做简单演示。真实项目建议保留原始字段，同时增加 normalized_time 字段。
try:
dt = date_parser.parse(text, fuzzy=True)
return dt.strftime("%Y-%m-%d %H:%M:%S")
except Exception:
return None

注意，不建议直接覆盖原始时间。更稳的做法是保留两个字段：

• time_raw：原始时间文本。
• time_normalized：程序解析后的标准时间。

因为自动解析可能出错，原始字段是回溯依据。

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

本文完成了一个完整的“国家公园活动公告分页”采集项目。我们从需求拆解开始，明确了目标字段：活动名、时间、地点、主办方、报名说明。然后按工程层次实现了请求层、解析层、存储层和主入口，并用本地 demo 站点保证代码可以真实运行。

这套项目的重点不在于“几行代码把网页扒下来”，而在于把采集过程做得可靠：

• 请求层有 headers、timeout、Session、重试和 robots.txt 检查。
• 解析层有列表页链接提取、详情页字段抽取和字段缺失容错。
• 存储层支持 CSV 和 JSONL，并通过 URL 做去重。
• 运行层有日志、输出路径和本地演示站点。
• 排错部分覆盖了 403、429、动态空壳、选择器变化和乱码问题。

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

第一，把本地 demo 换成真实站点。重点改 BASE_URL 和 config.py 里的选择器。

第二，增加 SQLite 或 MySQL 存储。数据量变大后，数据库比 CSV 更适合查询和去重。

第三，根据站点类型升级技术方案。如果页面是动态渲染，先找公开接口；如果接口不可用，再考虑 Playwright；如果采集规模扩大，再迁移到 Scrapy。

最后说一句个人感受：公告类爬虫看似朴素，但特别适合练基本功。它不像反爬对抗那样刺激，也不像大规模采集那样复杂，却能训练你对页面结构、字段容错、数据质量和工程边界的敏感度。把这种项目写稳，后面做更复杂的数据采集会顺很多。

附录：完整文件清单

为了方便复制，这里再整理一遍完整文件。

requirements.txt

requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.3.0
jsonlines==4.0.0
python-dateutil==2.9.0.post0

config.py

from pathlib import Path
import os

BASE_DIR = Path(__file__).resolve().parent

BASE_URL = os.getenv(
"PARK_EVENTS_BASE_URL",
"http://127.0.0.1:8000/events/page-{page}.html"
)

MAX_PAGES = int(os.getenv("PARK_EVENTS_MAX_PAGES", "5"))

MIN_DELAY = float(os.getenv("PARK_EVENTS_MIN_DELAY", "1.0"))
MAX_DELAY = float(os.getenv("PARK_EVENTS_MAX_DELAY", "2.5"))

CONNECT_TIMEOUT = 5
READ_TIMEOUT = 20

OUTPUT_DIR = BASE_DIR / "output"
OUTPUT_DIR.mkdir(exist_ok=True)

CSV_PATH = OUTPUT_DIR / "park_events.csv"
JSONL_PATH = OUTPUT_DIR / "park_events.jsonl"

LOG_DIR = BASE_DIR / "logs"
LOG_DIR.mkdir(exist_ok=True)
LOG_PATH = LOG_DIR / "spider.log"

LIST_ITEM_SELECTOR = ".event-card"
DETAIL_LINK_SELECTOR = "a.event-title"
NEXT_PAGE_SELECTOR = "a.next"

DETAIL_TITLE_SELECTOR = "h1"
DETAIL_CONTENT_SELECTOR = ".article-content"

FIELD_LABELS = {
"time": ["活动时间", "时间", "举办时间"],
"location": ["活动地点", "地点", "集合地点"],
"organizer": ["主办方", "主办单位", "组织单位"],
"registration": ["报名说明", "报名方式", "如何报名", "报名须知"],
}

models.py

from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional
import hashlib

@dataclass
class EventItem:
name: str
time: Optional[str]
location: Optional[str]
organizer: Optional[str]
registration: Optional[str]
detail_url: str
summary: Optional[str]
crawled_at: str

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

def content_hash(self) –> str:
raw = "|".join([
self.name or "",
self.time or "",
self.location or "",
self.organizer or "",
self.registration or "",
self.detail_url or "",
])
return hashlib.sha256(raw.encode("utf-8")).hexdigest()

fetcher.py

import logging
import random
import time
from urllib.parse import urlparse
from urllib.robotparser import RobotFileParser

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from config import (
CONNECT_TIMEOUT,
READ_TIMEOUT,
MIN_DELAY,
MAX_DELAY,
)

class Fetcher:
def __init__(self, user_agent: str = None):
self.logger = logging.getLogger(self.__class__.__name__)
self.session = requests.Session()

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

self.session.headers.update({
"User-Agent": self.user_agent,
"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",
"Connection": "keep-alive",
})

retry = Retry(
total=3,
connect=3,
read=3,
status=3,
backoff_factor=1.2,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET"],
raise_on_status=False,
)

adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)

self.robot_cache = {}

def sleep_polite(self):
seconds = random.uniform(MIN_DELAY, MAX_DELAY)
self.logger.debug("Sleep %.2f seconds before next request", seconds)
time.sleep(seconds)

def _robots_url(self, url: str) –> str:
parsed = urlparse(url)
return f"{parsed.scheme}://{parsed.netloc}/robots.txt"

def allowed_by_robots(self, url: str) –> bool:
parsed = urlparse(url)

if parsed.hostname in {"127.0.0.1", "localhost"}:
return True

robots_url = self._robots_url(url)

if robots_url not in self.robot_cache:
rp = RobotFileParser()
rp.set_url(robots_url)
try:
rp.read()
self.robot_cache[robots_url] = rp
except Exception as exc:
self.logger.warning("Read robots.txt failed: %s, error=%s", robots_url, exc)
return True

rp = self.robot_cache[robots_url]
return rp.can_fetch(self.user_agent, url)

def get(self, url: str, referer: str = None) –> str:
if not self.allowed_by_robots(url):
raise PermissionError(f"Blocked by robots.txt: {url}")

headers = {}
if referer:
headers["Referer"] = referer

self.sleep_polite()

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

response = self.session.get(
url,
headers=headers,
timeout=(CONNECT_TIMEOUT, READ_TIMEOUT),
)

if response.status_code == 403:
self.logger.warning("403 Forbidden: %s", url)
elif response.status_code == 429:
self.logger.warning("429 Too Many Requests: %s", url)

response.raise_for_status()

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

return response.text

parser.py

import re
from datetime import datetime
from typing import List, Optional
from urllib.parse import urljoin

from bs4 import BeautifulSoup

from config import (
LIST_ITEM_SELECTOR,
DETAIL_LINK_SELECTOR,
NEXT_PAGE_SELECTOR,
DETAIL_TITLE_SELECTOR,
DETAIL_CONTENT_SELECTOR,
FIELD_LABELS,
)
from models import EventItem

def clean_text(text: Optional[str]) –> Optional[str]:
if text is None:
return None
text = re.sub(r"\\s+", " ", text)
text = text.replace("\\u3000", " ")
text = text.strip(" ：:;；\\n\\t")
return text or None

def compact_text(text: Optional[str], limit: int = 140) –> Optional[str]:
text = clean_text(text)
if not text:
return None
return text[:limit] + ("…" if len(text) > limit else "")

def soupify(html: str) –> BeautifulSoup:
return BeautifulSoup(html, "lxml")

class EventParser:
def parse_list_page(self, html: str, base_url: str) –> List[str]:
soup = soupify(html)
links = []

cards = soup.select(LIST_ITEM_SELECTOR)

if not cards:
for a in soup.select(DETAIL_LINK_SELECTOR):
href = a.get("href")
if href:
links.append(urljoin(base_url, href))
return list(dict.fromkeys(links))

for card in cards:
a = card.select_one(DETAIL_LINK_SELECTOR)
if not a:
continue

href = a.get("href")
if not href:
continue

detail_url = urljoin(base_url, href)
links.append(detail_url)

return list(dict.fromkeys(links))

def parse_next_page(self, html: str, base_url: str) –> Optional[str]:
soup = soupify(html)
a = soup.select_one(NEXT_PAGE_SELECTOR)
if a and a.get("href"):
return urljoin(base_url, a["href"])
return None

def parse_detail_page(self, html: str, detail_url: str) –> EventItem:
soup = soupify(html)

title = self._extract_title(soup)
content_node = soup.select_one(DETAIL_CONTENT_SELECTOR)

if content_node:
content_text = content_node.get_text("\\n", strip=True)
else:
body = soup.select_one("body")
content_text = body.get_text("\\n", strip=True) if body else soup.get_text("\\n", strip=True)

content_text = clean_text(content_text) or ""

field_map = self._extract_fields_from_content(content_text)

item = EventItem(
name=title or self._guess_title_from_content(content_text) or "未命名活动",
time=field_map.get("time"),
location=field_map.get("location"),
organizer=field_map.get("organizer"),
registration=field_map.get("registration"),
detail_url=detail_url,
summary=compact_text(content_text, limit=180),
crawled_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
)

return item

def _extract_title(self, soup: BeautifulSoup) –> Optional[str]:
title_node = soup.select_one(DETAIL_TITLE_SELECTOR)
if title_node:
return clean_text(title_node.get_text(" ", strip=True))

if soup.title:
return clean_text(soup.title.get_text(" ", strip=True))

return None

def _guess_title_from_content(self, content: str) –> Optional[str]:
if not content:
return None

lines = [clean_text(x) for x in content.split("\\n")]
lines = [x for x in lines if x]

if not lines:
return None

return lines[0][:80]

def _extract_fields_from_content(self, content: str) –> dict:
result = {
"time": None,
"location": None,
"organizer": None,
"registration": None,
}

if not content:
return result

lines = re.split(r"[\\n\\r]+|(?<=。)", content)
lines = [clean_text(line) for line in lines]
lines = [line for line in lines if line]

for key, labels in FIELD_LABELS.items():
value = self._find_value_by_labels(lines, labels)
if value:
result[key] = value

if not result["registration"]:
result["registration"] = self._guess_registration(content)

return result

def _find_value_by_labels(self, lines: List[str], labels: List[str]) –> Optional[str]:
for line in lines:
for label in labels:
pattern = rf"^{re.escape(label)}\\s*[:：]\\s*(.+)$"
m = re.search(pattern, line)
if m:
return clean_text(m.group(1))

pattern2 = rf"{re.escape(label)}\\s*[:：]\\s*([^。；;]+)"
m2 = re.search(pattern2, line)
if m2:
return clean_text(m2.group(1))

return None

def _guess_registration(self, content: str) –> Optional[str]:
sentences = re.split(r"[。！？!?]\\s*", content)
candidates = []
for sentence in sentences:
if any(word in sentence for word in ["报名", "预约", "名额", "参加方式", "二维码"]):
candidates.append(clean_text(sentence))

if not candidates:
return None

return "；".join([x for x in candidates if x][:2])

storage.py

import csv
import jsonlines
from pathlib import Path
from typing import Iterable, List, Set

from models import EventItem

FIELDNAMES = [
"name",
"time",
"location",
"organizer",
"registration",
"detail_url",
"summary",
"crawled_at",
]

class EventStorage:
def __init__(self, csv_path: Path, jsonl_path: Path):
self.csv_path = csv_path
self.jsonl_path = jsonl_path

self.csv_path.parent.mkdir(parents=True, exist_ok=True)
self.jsonl_path.parent.mkdir(parents=True, exist_ok=True)

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

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

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

return seen

def save_csv(self, items: Iterable[EventItem], append: bool = True) –> int:
items = list(items)
if not items:
return 0

file_exists = self.csv_path.exists()
mode = "a" if append else "w"

with self.csv_path.open(mode, encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=FIELDNAMES)

if not file_exists or not append:
writer.writeheader()

for item in items:
writer.writerow(item.to_dict())

return len(items)

def save_jsonl(self, items: Iterable[EventItem], append: bool = True) –> int:
items = list(items)
if not items:
return 0

mode = "a" if append else "w"

with jsonlines.open(self.jsonl_path, mode=mode) as writer:
for item in items:
writer.write(item.to_dict())

return len(items)

def save_all(self, items: List[EventItem]) –> int:
count_csv = self.save_csv(items, append=True)
self.save_jsonl(items, append=True)
return count_csv

main.py

import logging

from config import (
BASE_URL,
MAX_PAGES,
CSV_PATH,
JSONL_PATH,
LOG_PATH,
)
from fetcher import Fetcher
from parser import EventParser
from storage import EventStorage

def setup_logging():
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
handlers=[
logging.FileHandler(LOG_PATH, encoding="utf-8"),
logging.StreamHandler(),
],
)

def build_list_url(page: int) –> str:
if "{page}" in BASE_URL:
return BASE_URL.format(page=page)
if page == 1:
return BASE_URL
separator = "&" if "?" in BASE_URL else "?"
return f"{BASE_URL}{separator}page={page}"

def run():
setup_logging()

logger = logging.getLogger("park_events_spider")

fetcher = Fetcher()
parser = EventParser()
storage = EventStorage(CSV_PATH, JSONL_PATH)

seen_urls = storage.load_seen_urls()
logger.info("Loaded seen urls: %d", len(seen_urls))

all_new_items = []
current_url = None

for page in range(1, MAX_PAGES + 1):
list_url = current_url or build_list_url(page)

try:
html = fetcher.get(list_url)
except Exception as exc:
logger.exception("Fetch list page failed: page=%s url=%s error=%s", page, list_url, exc)
break

detail_urls = parser.parse_list_page(html, list_url)

if not detail_urls:
logger.info("No detail urls found, stop. page=%s url=%s", page, list_url)
break

logger.info("Page %s found %d detail urls", page, len(detail_urls))

page_items = []

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.get(detail_url, referer=list_url)
item = parser.parse_detail_page(detail_html, detail_url)
except Exception as exc:
logger.exception("Fetch or parse detail failed: url=%s error=%s", detail_url, exc)
continue

seen_urls.add(detail_url)
page_items.append(item)

logger.info(
"Parsed event: name=%s time=%s location=%s",
item.name,
item.time,
item.location,
)

if page_items:
saved_count = storage.save_all(page_items)
all_new_items.extend(page_items)
logger.info("Saved %d items from page %s", saved_count, page)

next_url = parser.parse_next_page(html, list_url)

if "{page}" not in BASE_URL:
if not next_url:
logger.info("No next page found, stop.")
break
current_url = next_url

logger.info("Spider finished. New items: %d", len(all_new_items))
logger.info("CSV output: %s", CSV_PATH)
logger.info("JSONL output: %s", JSONL_PATH)

if __name__ == "__main__":
run()

tools/make_demo_site.py

from pathlib import Path

BASE_DIR = Path(__file__).resolve().parents[1]
SITE_DIR = BASE_DIR / "demo_site"
EVENTS_DIR = SITE_DIR / "events"

DETAIL_TEMPLATE = """<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>{title}</title>
</head>
<body>
<main class="page">
<article class="article-content">
<h1>{title}</h1>
<p>活动时间：{time}</p>
<p>活动地点：{location}</p>
<p>主办方：{organizer}</p>
<p>报名说明：{registration}</p>
<p>{intro}</p>
<p>请参与者根据天气情况准备舒适鞋服，并遵守现场工作人员指引。</p>
</article>
</main>
</body>
</html>
"""

LIST_TEMPLATE = """<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>国家公园活动公告 – 第 {page} 页</title>
</head>
<body>
<section class="event-list">
<h1>国家公园活动公告</h1>
{cards}
</section>
<nav class="pagination">
{next_link}
</nav>
</body>
</html>
"""

CARD_TEMPLATE = """
<div class="event-card">
<h2><a class="event-title" href="{href}">{title}</a></h2>
<p class="event-date">{time}</p>
<p class="event-desc">{desc}</p>
</div>
"""

EVENTS = [
{
"title": "夜观昆虫：国家公园夏季自然观察活动",
"time": "2026年6月15日 19:30-21:00",
"location": "青溪国家公园访客中心集合",
"organizer": "青溪国家公园管理处",
"registration": "请通过官网活动预约入口报名，名额30人，额满即止。",
"intro": "本次活动面向自然观察爱好者，带领参与者认识夜间昆虫和栖息地环境。",
},
{
"title": "高山植物识别公益讲解",
"time": "2026年6月18日 09:00-11:30",
"location": "云岭国家公园科普步道",
"organizer": "云岭国家公园自然教育中心",
"registration": "无需缴费，需提前一天电话预约，团队人数不超过20人。",
"intro": "讲解员将沿科普步道介绍常见高山植物的形态特征和生态价值。",
},
{
"title": "湿地候鸟监测志愿者体验日",
"time": "2026年6月22日 07:30-10:30",
"location": "湖湾湿地观鸟平台",
"organizer": "湖湾国家公园志愿服务队",
"registration": "报名时请填写姓名、联系方式和是否自备望远镜。",
"intro": "活动将介绍基础鸟类识别方法，并体验标准化监测记录流程。",
},
{
"title": "亲子自然笔记工作坊",
"time": "2026年6月28日 14:00-16:30",
"location": "松谷国家公园自然教室",
"organizer": "松谷国家公园宣教科",
"registration": "适合8岁以上儿童家庭参加，每组家庭最多2名成人陪同。",
"intro": "老师会带领孩子观察叶片、果实和昆虫痕迹，并完成一页自然笔记。",
},
{
"title": "国家公园巡护员的一天主题分享",
"time": "2026年7月3日 10:00-11:30",
"location": "青溪国家公园报告厅",
"organizer": "青溪国家公园管理处",
"registration": "现场座位有限，建议提前在线预约。",
"intro": "一线巡护员将分享巡护路线、野外记录和公众参与保护的方式。",
},
{
"title": "溪流生态观察与水质小实验",
"time": "2026年7月6日 09:30-12:00",
"location": "石桥溪生态观察点",
"organizer": "青溪国家公园自然教育团队",
"registration": "活动需穿防滑鞋，不建议6岁以下儿童参加。",
"intro": "活动通过简易水质测试和底栖动物观察，认识溪流生态系统。",
},
]

def build():
EVENTS_DIR.mkdir(parents=True, exist_ok=True)

for idx, event in enumerate(EVENTS, start=1):
detail_html = DETAIL_TEMPLATE.format(**event)
detail_path = EVENTS_DIR / f"detail-{idx}.html"
detail_path.write_text(detail_html, encoding="utf-8")

page_size = 3
pages = [EVENTS[i:i + page_size] for i in range(0, len(EVENTS), page_size)]

for page_index, page_events in enumerate(pages, start=1):
cards = []
for event in page_events:
idx = EVENTS.index(event) + 1
card = CARD_TEMPLATE.format(
href=f"detail-{idx}.html",
title=event["title"],
time=event["time"],
desc=event["intro"],
)
cards.append(card)

next_link = ""
if page_index < len(pages):
next_link = f'<a class="next" href="page-{page_index + 1}.html">下一页</a>'

list_html = LIST_TEMPLATE.format(
page=page_index,
cards="\\n".join(cards),
next_link=next_link,
)

page_path = EVENTS_DIR / f"page-{page_index}.html"
page_path.write_text(list_html, encoding="utf-8")

print(f"Demo site generated at: {SITE_DIR}")

if __name__ == "__main__":
build()

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

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

✅ 免责声明

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

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

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