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

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

这篇文章要做的事很直接：采集牧场、牧区相关公开项目页面，用 requests + BeautifulSoup + SQLite 做项目制数据归档，最终输出一份可增量更新的项目数据库和 CSV 文件。

读完之后，你至少能拿到三样东西：

第一，知道“公开项目归档”这种数据该怎么拆成列表页、详情页、字段、唯一键和增量逻辑。

第二，拿到一套可运行的 Python 爬虫代码，默认带本地示例页面，不依赖真实网站也能跑通。

第三，明白遇到 403、429、动态渲染、字段缺失、页面结构变动时，应该从哪里排查，而不是只盯着一行报错发呆。

本文主题是“牧场/牧区公开项目归档”，字段包括：

字段	说明
项目名	项目的公开名称
地区	项目所属区域、牧区、县域或片区
牲畜类型	牛、羊、牦牛、马、骆驼等
周期	项目执行周期、建设周期、服务周期
主管单位	发布单位、主管部门、组织单位或项目承担单位
链接	详情页 URL

文章的定位不是“教你硬闯某个网站”，而是把一个现实里很常见的项目制公开信息，整理成稳定的数据采集、清洗、存储和增量归档流程。站点没有固定指定时，我会采用“静态列表页 + 详情页”的通用方案，并提供本地 demo 页面保证代码可运行。换到真实站点时，只需要替换入口 URL 和选择器。

1️⃣ 摘要（Abstract）

本文围绕“牧场/牧区公开项目归档”这个采集任务，使用 Python 的 requests、BeautifulSoup 和 SQLite 搭建一套从公开网页采集项目数据、解析详情字段、去重入库并导出 CSV 的增量采集器。

读完本文，你可以获得：

1. 一套适合项目归档类数据的爬虫工程结构。
2. 一个可运行的 Fetcher、Parser、Storage、Pipeline 分层实现。
3. 一个清晰的增量策略：URL 唯一、内容 hash 判断变化、首次发现和最近更新时间记录。

我个人很喜欢做这种“小而稳”的数据归档工具。它没有花哨的界面，也不追求一秒抓几万页，但它解决的是很真实的问题：公开信息分散在网页上，人工复制容易漏、容易乱、容易重复，而脚本可以把它们整理成结构化资产。

---

2️⃣ 背景与需求（Why）

2.1 为什么要爬牧场/牧区公开项目？

公开项目页面通常有一个共同特点：信息不一定难找，但分散、重复、格式不统一。

比如某些站点会按年份发布牧区建设项目、草畜平衡项目、畜牧养殖示范项目、牧场改良项目、疫病防控项目、养殖基础设施项目等。人看一两页还行，如果要长期跟踪，就会遇到几个问题：

第一，信息分散在列表页和详情页。列表页一般只有标题、日期、摘要，真正有用的字段藏在详情页正文、表格或者附件说明里。

第二，项目会持续新增。今天抓到 200 条，下个月可能新增 30 条。如果每次全量复制，非常浪费，也容易重复。

第三，字段表达不统一。有的页面写“主管单位”，有的写“组织单位”，有的写“实施单位”，还有的只在正文里出现“由某某单位负责实施”。这就要求解析层要有容错。

第四，项目数据后续通常要进入分析流程。比如按地区统计项目数量，按牲畜类型统计项目分布，按周期筛选长期项目，或者按主管单位做归档台账。

所以，这类任务的核心并不是“抓网页”三个字，而是四件事：

采集公开页面 → 解析项目字段 → 清洗统一格式 → 增量入库归档

2.2 目标字段

本次采集字段固定为：

字段名	英文字段	说明
项目名	project_name	项目公开名称
地区	region	项目所属地区、牧区、县域或片区
牲畜类型	livestock_type	牛、羊、牦牛、马、骆驼等
周期	cycle	项目周期、建设周期、服务周期
主管单位	authority	主管单位、发布单位、组织单位、实施单位
链接	url	项目详情页链接

为了做增量，还会额外保存几个系统字段：

字段名	说明
source_hash	页面正文 hash，用来判断内容是否变化
first_seen_at	首次采集时间
last_seen_at	最近采集或更新时间
created_at	入库时间
updated_at	最近更新入库时间

这些额外字段不是展示重点，但对长期归档很有用。尤其是 source_hash，它能帮我们判断同一个 URL 的内容是否改过。

---

3️⃣ 合规与注意事项

3.1 robots.txt 是什么？

robots.txt 是网站放在根路径下的一份爬虫访问说明文件，例如：

https://example.com/robots.txt

它会告诉爬虫哪些路径可以访问，哪些路径不建议访问。严格来说，robots.txt 本身不是身份认证系统，也不是权限系统，但作为技术采集者，应该尊重站点的访问规则。

本文代码里会提供 --respect-robots 参数。打开后，脚本会在请求前读取目标站点的 robots.txt，再判断当前 URL 是否允许当前 User-Agent 访问。

实际项目里，我建议至少做到这几点：

1. 先看目标站点是否有 robots.txt。
2. 不采集明确禁止抓取的路径。
3. 不绕过登录、验证码、付费墙。
4. 不对站点制造高频访问压力。

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

公开页面采集不是压测。对于项目归档类任务，通常不需要极端并发。

本文默认采用串行抓取，并在每次请求之间加入随机等待：

sleep_min = 0.7
sleep_max = 1.5

真实项目里，如果站点规模不大，几百页、几千页都可以慢慢跑。我的习惯是先稳定，再谈速度。尤其是归档任务，不是秒级业务，没必要把自己写成“疑似异常流量”。

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

本文只讨论公开项目页面中已经展示的项目制信息，比如项目名、地区、牲畜类型、周期、主管单位和链接。

不建议采集以下内容：

个人手机号
身份证号
住址
账号信息
登录后才可见的数据
付费后才可见的数据
明确禁止自动访问的数据

如果页面需要登录、验证码、付费权限或特殊授权，那就不应该用爬虫绕过去。更稳妥的做法是联系数据发布方，申请接口、下载文件或授权访问。

3.4 中性采集，不做越界判断

本文不会评价项目本身，也不会做涉政分析。我们只从工程角度讨论如何把公开网页中的项目数据整理为结构化表格。数据采集只是工具，边界要清楚。

---

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

4.1 本文属于哪种页面采集？

网页采集通常分三类：

类型	特点	常见工具
静态页面	HTML 里直接有数据	requests、BeautifulSoup、lxml
动态页面	数据由 JS 渲染出来	Playwright、Selenium、接口分析
API 接口	页面请求 JSON 接口	requests、httpx

本文默认采用第一种：静态公开列表页 + 详情页。

原因很简单：很多项目公开页面本质上还是传统 HTML 页面。列表页里有项目链接，详情页里有正文、表格、字段描述。这种场景用 requests + BeautifulSoup 足够稳，也更容易部署。

如果你后面遇到动态页面，比如浏览器里能看到数据，但 requests.get() 抓到的是空壳，那再切到 Playwright 或分析接口。

4.2 整体流程

本文的流程可以写成：

入口列表页
   ↓
请求 Fetcher
   ↓
解析列表页，提取详情链接
   ↓
逐个请求详情页
   ↓
解析详情字段
   ↓
清洗字段、补空值、计算 hash
   ↓
SQLite upsert
   ↓
导出 CSV

如果用 Mermaid 表达，大概是这样：

4.3 为什么选 requests + BeautifulSoup + SQLite？

我这里不选特别重的框架，主要基于几个考虑。

requests 适合处理普通 HTTP 请求。它简单、稳定、生态成熟，对 headers、timeout、session、重试都能很好控制。

BeautifulSoup 适合结构不算特别规整的页面。项目公开页面经常有表格、段落、换行、全角冒号、不同标签混在一起的情况，BeautifulSoup 的容错比纯 XPath 写起来更舒服。

SQLite 适合本地归档。它不需要额外部署数据库，单文件存储，支持唯一约束和更新操作。对于个人项目、小团队归档、每日定时采集，SQLite 已经够用。

如果数据量继续变大，可以再迁移到 MySQL、PostgreSQL 或者 Elasticsearch。第一版没必要一上来就堆复杂度。

---

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

5.1 Python 版本

建议使用：

Python 3.10+

我个人会优先选择 Python 3.11。原因不是它一定能让爬虫快多少，而是类型提示、标准库和异常信息都更舒服。

5.2 安装依赖

新建项目目录：

mkdir ranch_project_archive
cd ranch_project_archive

创建虚拟环境：

python -m venv .venv

激活虚拟环境。

Windows：

.venv\Scripts\activate

macOS / Linux：

source .venv/bin/activate

安装依赖：

pip install requests beautifulsoup4 lxml

生成 requirements.txt：

requests>=2.31.0
beautifulsoup4>=4.12.0
lxml>=5.0.0

5.3 推荐项目结构

本文项目结构如下：

ranch_project_archive/
├── crawler/
│   ├── __init__.py
│   ├── config.py
│   ├── fetcher.py
│   ├── parser.py
│   ├── storage.py
│   ├── pipeline.py
│   └── utils.py
├── examples/
│   ├── __init__.py
│   └── demo_site.py
├── data/
│   ├── raw/
│   ├── ranch_projects.sqlite3
│   └── ranch_projects.csv
├── main.py
└── requirements.txt

这里我把爬虫拆成了几层：

文件	作用
config.py	配置入口、选择器、请求参数
fetcher.py	请求层，负责 headers、timeout、重试、robots
parser.py	解析层，负责列表链接和详情字段
storage.py	存储层，负责 SQLite 入库和 CSV 导出
pipeline.py	流程编排
utils.py	通用工具
demo_site.py	本地示例页面，保证代码可运行
main.py	命令行入口

我建议你不要把所有代码塞进一个 spider.py。短期看省事，后期一定乱。尤其是项目制归档要长期跑，Fetcher、Parser、Storage 分开之后，任何一层出问题都更容易定位。

---

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

请求层要解决的不是“能不能 GET 一下”，而是几个非常实际的问题：

headers 怎么写
timeout 怎么控制
失败了怎么重试
429 怎么退避
要不要 session
要不要检查 robots.txt

6.1 配置文件：crawler/config.py

# crawler/config.py
from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class SiteConfig:
    """
    采集配置。

    默认配置指向本地 demo 服务：
    http://127.0.0.1:8765/list.html

    换成真实站点时，通常只需要改：
    - base_url
    - list_paths
    - list_link_selector
    - title_selector
    - detail_container_selector
    """

    base_url: str = "http://127.0.0.1:8765"
    list_paths: list[str] = field(default_factory=lambda: ["/list.html"])

    list_link_selector: str = "a.project-link"
    title_selector: str = "h1"
    detail_container_selector: str = "main"

    user_agent: str = (
        "Mozilla/5.0 (compatible; RanchArchiveBot/1.0; "
        "+https://example.local/bot-info)"
    )

    request_timeout: int = 10
    max_retries: int = 3
    backoff_base: float = 1.2

    sleep_min: float = 0.7
    sleep_max: float = 1.5

    respect_robots: bool = False

    data_dir: Path = Path("data")
    raw_dir: Path = Path("data/raw")
    db_path: Path = Path("data/ranch_projects.sqlite3")
    csv_path: Path = Path("data/ranch_projects.csv")

    field_aliases: dict[str, list[str]] = field(
        default_factory=lambda: {
            "region": [
                "地区",
                "所属地区",
                "项目地区",
                "所在地区",
                "牧区",
                "区域",
            ],
            "livestock_type": [
                "牲畜类型",
                "畜种",
                "养殖类型",
                "畜牧类型",
                "主要牲畜",
                "适用畜种",
            ],
            "cycle": [
                "周期",
                "项目周期",
                "建设周期",
                "执行周期",
                "服务周期",
                "实施周期",
            ],
            "authority": [
                "主管单位",
                "发布单位",
                "组织单位",
                "实施单位",
                "承担单位",
                "管理单位",
            ],
        }
    )

这个配置文件有两个特点：

第一，默认能跑本地 demo。你不用先找真实网站，也不用担心线上页面改版导致示例代码第一步就失败。

第二，字段别名放在配置里。因为现实里的页面不会都乖乖写“牲畜类型”。有些写“畜种”，有些写“主要牲畜”，还有些写“适用畜种”。别名表可以让解析层更耐用。

6.2 工具函数：crawler/utils.py

# crawler/utils.py
from __future__ import annotations

import hashlib
import re
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse


def now_iso() -> str:
    """返回 UTC ISO 时间字符串。"""
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def sha256_text(text: str) -> str:
    """计算文本 sha256。"""
    return hashlib.sha256(text.encode("utf-8", errors="ignore")).hexdigest()


def normalize_space(text: str | None) -> str:
    """压缩空白字符。"""
    if not text:
        return ""
    return re.sub(r"\s+", " ", text).strip()


def normalize_label(text: str | None) -> str:
    """
    规范化字段名：
    - 去空格
    - 去全角/半角冒号
    - 去常见标点
    """
    if not text:
        return ""
    text = normalize_space(text)
    text = text.replace("：", ":")
    text = text.strip(" :：\t\r\n")
    return text


def safe_filename_from_url(url: str, suffix: str = ".html") -> str:
    """
    根据 URL 生成安全文件名，用于保存 raw html。
    """
    parsed = urlparse(url)
    raw = f"{parsed.netloc}{parsed.path}"
    raw = raw.strip("/") or "index"
    raw = re.sub(r"[^0-9a-zA-Z\u4e00-\u9fff._-]+", "_", raw)
    if not raw.endswith(suffix):
        raw += suffix
    return raw


def ensure_dir(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True)

这些小函数看起来不起眼，但后期省很多事。比如 normalize_space() 可以处理网页里的各种换行和空格，safe_filename_from_url() 可以把原始页面保存下来，方便排错。

6.3 请求实现：crawler/fetcher.py

# crawler/fetcher.py
from __future__ import annotations

import random
import time
from dataclasses import dataclass
from typing import Optional
from urllib.parse import urljoin, urlparse
from urllib.robotparser import RobotFileParser

import requests

from crawler.config import SiteConfig


class FetchError(RuntimeError):
    """请求失败时抛出的异常。"""


@dataclass
class FetchResult:
    url: str
    status_code: int
    text: str
    elapsed: float


class Fetcher:
    """
    请求层。

    负责：
    - 复用 requests.Session
    - 设置 UA、Referer
    - timeout
    - 重试与退避
    - 可选 robots.txt 检查
    """

    def __init__(self, config: SiteConfig):
        self.config = config
        self.session = requests.Session()
        self.session.headers.update(
            {
                "User-Agent": self.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",
            }
        )
        self._robots_cache: dict[str, RobotFileParser] = {}

    def build_url(self, path_or_url: str) -> str:
        return urljoin(self.config.base_url, path_or_url)

    def can_fetch(self, url: str) -> bool:
        """
        根据 robots.txt 判断是否允许抓取。

        注意：
        - robots 获取失败时，不代表一定禁止，也不代表一定允许。
        - 本示例按“读取不到则默认允许”处理。
        - 严肃生产环境可以改成“读取不到则跳过”。
        """
        parsed = urlparse(url)
        root = f"{parsed.scheme}://{parsed.netloc}"
        robots_url = urljoin(root, "/robots.txt")

        if robots_url not in self._robots_cache:
            rp = RobotFileParser()
            rp.set_url(robots_url)
            try:
                rp.read()
            except Exception:
                return True
            self._robots_cache[robots_url] = rp

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

    def polite_sleep(self) -> None:
        delay = random.uniform(self.config.sleep_min, self.config.sleep_max)
        time.sleep(delay)

    def get(self, url: str, referer: Optional[str] = None) -> FetchResult:
        """
        GET 请求，带重试和指数退避。

        对 429、5xx 做重试；
        对 403、404 这类通常不盲目重试。
        """
        if self.config.respect_robots and not self.can_fetch(url):
            raise FetchError(f"Blocked by robots.txt: {url}")

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

        last_error: Exception | None = None

        for attempt in range(1, self.config.max_retries + 1):
            start = time.perf_counter()
            try:
                response = self.session.get(
                    url,
                    headers=headers,
                    timeout=self.config.request_timeout,
                )
                elapsed = time.perf_counter() - start

                if response.status_code in {429, 500, 502, 503, 504}:
                    raise FetchError(
                        f"Retryable status {response.status_code} for {url}"
                    )

                response.raise_for_status()

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

                return FetchResult(
                    url=response.url,
                    status_code=response.status_code,
                    text=response.text,
                    elapsed=elapsed,
                )

            except Exception as exc:
                last_error = exc

                if attempt >= self.config.max_retries:
                    break

                sleep_seconds = (
                    self.config.backoff_base * (2 ** (attempt - 1))
                    + random.uniform(0, 0.5)
                )
                time.sleep(sleep_seconds)

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

6.4 请求层重点说明

headers

这里设置了比较正常的请求头：

"User-Agent": "...RanchArchiveBot/1.0..."
"Accept": "text/html..."
"Accept-Language": "zh-CN..."

User-Agent 不建议伪装成特别奇怪的东西。真实项目里，最好写清楚自己的采集器名称和说明页。本文为了演示，使用了 RanchArchiveBot/1.0。

Referer 只在请求详情页时传入列表页 URL，不做过度伪装。

timeout

不要写没有 timeout 的请求：

requests.get(url)

如果站点卡住，脚本也会卡住。本文配置了：

request_timeout = 10

这个值可以按网络环境调整。内网或慢站点可以放到 20 秒，普通公开网页 10 秒通常够用。

session / cookie

本文用了 requests.Session()，主要是为了复用连接和统一 headers。

如果真实站点不需要登录，就不要手动处理 cookie。公开页面采集一般不应该依赖登录态。如果某个站点要求登录才能访问，那就不属于本文讨论范围。

失败处理：重试和退避

请求失败不能立刻疯狂重试。本文对 429 和 5xx 做指数退避：

sleep_seconds = backoff_base * (2 ** (attempt - 1)) + jitter

也就是说第一次失败等短一点，第二次失败等更久一点。这样能减少对目标站点的压力，也能提高临时网络波动下的成功率。

---

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

解析层是这类任务最容易脏的地方。真实网页经常出现：

字段写在 table 里
字段写在 dl/dd 里
字段写在 p 标签里
字段名有全角冒号
字段名换行
有些字段缺失
详情链接是相对路径

所以解析层要尽量“稳”，不要假设页面永远整齐。

7.1 数据模型和解析代码：crawler/parser.py

# crawler/parser.py
from __future__ import annotations

import re
from dataclasses import dataclass
from urllib.parse import urljoin

from bs4 import BeautifulSoup, Tag

from crawler.config import SiteConfig
from crawler.utils import normalize_label, normalize_space, sha256_text


@dataclass
class RanchProject:
    project_name: str
    region: str
    livestock_type: str
    cycle: str
    authority: str
    url: str
    source_hash: str


class Parser:
    """
    解析层。

    负责：
    - 从列表页提取详情链接
    - 从详情页抽取项目字段
    - 处理字段缺失
    - 计算内容 hash
    """

    def __init__(self, config: SiteConfig):
        self.config = config
        self.alias_to_field = self._build_alias_to_field(config.field_aliases)

    @staticmethod
    def _build_alias_to_field(field_aliases: dict[str, list[str]]) -> dict[str, str]:
        result: dict[str, str] = {}
        for field, aliases in field_aliases.items():
            for alias in aliases:
                result[normalize_label(alias)] = field
        return result

    def parse_list_links(self, html: str, page_url: str) -> list[str]:
        """
        从列表页解析详情链接。

        默认选择器是 a.project-link。
        换真实站点时，可以在 config.py 修改 list_link_selector。
        """
        soup = BeautifulSoup(html, "lxml")
        links: list[str] = []

        for node in soup.select(self.config.list_link_selector):
            href = node.get("href")
            if not href:
                continue

            detail_url = urljoin(page_url, href)
            if detail_url not in links:
                links.append(detail_url)

        return links

    def parse_detail(self, html: str, url: str) -> RanchProject:
        soup = BeautifulSoup(html, "lxml")

        container = soup.select_one(self.config.detail_container_selector)
        if container is None:
            container = soup.body or soup

        title = self._extract_title(soup, container)
        pairs = self._extract_key_value_pairs(container)
        fallback_text = normalize_space(container.get_text(" ", strip=True))

        fields = {
            "region": "",
            "livestock_type": "",
            "cycle": "",
            "authority": "",
        }

        for label, value in pairs.items():
            normalized = normalize_label(label)
            target_field = self.alias_to_field.get(normalized)
            if target_field and not fields[target_field]:
                fields[target_field] = normalize_space(value)

        # 如果结构化字段没取到，再从正文里做轻量兜底。
        for field in fields:
            if not fields[field]:
                fields[field] = self._fallback_extract(field, fallback_text)

        # 项目名兜底：标题拿不到时，尝试从正文里找“项目名”
        if not title:
            title = self._extract_project_name_from_pairs(pairs) or "未命名项目"

        content_hash = sha256_text(fallback_text)

        return RanchProject(
            project_name=title,
            region=fields["region"],
            livestock_type=fields["livestock_type"],
            cycle=fields["cycle"],
            authority=fields["authority"],
            url=url,
            source_hash=content_hash,
        )

    def _extract_title(self, soup: BeautifulSoup, container: Tag) -> str:
        title_node = soup.select_one(self.config.title_selector)
        if title_node:
            return normalize_space(title_node.get_text(" ", strip=True))

        h1 = container.find("h1")
        if h1:
            return normalize_space(h1.get_text(" ", strip=True))

        h2 = container.find("h2")
        if h2:
            return normalize_space(h2.get_text(" ", strip=True))

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

        return ""

    def _extract_key_value_pairs(self, container: Tag) -> dict[str, str]:
        """
        尽量从 table、dl、p/li 文本中提取 key-value。

        支持格式：
        - <tr><th>地区</th><td>青岭牧区</td></tr>
        - <dt>地区</dt><dd>青岭牧区</dd>
        - <p>地区：青岭牧区</p>
        - <li>牲畜类型：牦牛、藏羊</li>
        """
        pairs: dict[str, str] = {}

        # table 结构
        for tr in container.select("tr"):
            cells = tr.find_all(["th", "td"])
            if len(cells) >= 2:
                key = normalize_label(cells[0].get_text(" ", strip=True))
                value = normalize_space(cells[1].get_text(" ", strip=True))
                if key and value:
                    pairs[key] = value

        # dl/dt/dd 结构
        for dl in container.select("dl"):
            children = [child for child in dl.children if isinstance(child, Tag)]
            for index, child in enumerate(children):
                if child.name != "dt":
                    continue
                if index + 1 >= len(children):
                    continue
                next_node = children[index + 1]
                if next_node.name != "dd":
                    continue

                key = normalize_label(child.get_text(" ", strip=True))
                value = normalize_space(next_node.get_text(" ", strip=True))
                if key and value:
                    pairs[key] = value

        # 段落 / 列表结构
        text_nodes = container.find_all(["p", "li", "div"])
        for node in text_nodes:
            text = normalize_space(node.get_text(" ", strip=True))
            if not text:
                continue

            parsed = self._split_key_value_text(text)
            if parsed:
                key, value = parsed
                if key and value and key not in pairs:
                    pairs[key] = value

        return pairs

    @staticmethod
    def _split_key_value_text(text: str) -> tuple[str, str] | None:
        """
        将 “字段：值” 或 “字段: 值” 拆开。
        为了避免误伤正文，只处理前半段较短的文本。
        """
        text = text.replace("：", ":")
        if ":" not in text:
            return None

        key, value = text.split(":", 1)
        key = normalize_label(key)
        value = normalize_space(value)

        if not key or not value:
            return None

        if len(key) > 20:
            return None

        return key, value

    def _extract_project_name_from_pairs(self, pairs: dict[str, str]) -> str:
        for key in ["项目名", "项目名称", "名称"]:
            if key in pairs:
                return normalize_space(pairs[key])
        return ""

    def _fallback_extract(self, field: str, text: str) -> str:
        """
        正文兜底抽取。

        这不是 NLP，只是轻量规则。
        对项目归档来说，够用、可控、容易维护。
        """
        patterns = {
            "region": [
                r"(?:地区|所属地区|所在地区|牧区|区域)[:：]\s*([^。；;，,\s]+)",
            ],
            "livestock_type": [
                r"(?:牲畜类型|畜种|养殖类型|主要牲畜|适用畜种)[:：]\s*([^。；;]+)",
            ],
            "cycle": [
                r"(?:周期|项目周期|建设周期|执行周期|服务周期|实施周期)[:：]\s*([^。；;]+)",
            ],
            "authority": [
                r"(?:主管单位|发布单位|组织单位|实施单位|承担单位|管理单位)[:：]\s*([^。；;]+)",
            ],
        }

        for pattern in patterns.get(field, []):
            match = re.search(pattern, text)
            if match:
                return normalize_space(match.group(1))

        return ""

7.2 列表页如何拿详情链接？

默认逻辑是：

for node in soup.select("a.project-link"):
    href = node.get("href")
    detail_url = urljoin(page_url, href)

这里一定要用 urljoin()。因为真实页面常见的链接有三种：

/project/1001.html
../detail/1001.html
https://example.com/project/1001.html

不用 urljoin()，你很容易把相对路径拼错。

7.3 详情页如何抽字段？

优先级如下：

table / dl / p / li 结构化字段
   ↓
正文正则兜底
   ↓
缺失字段留空

为什么缺失字段不直接报错？

因为公开项目页面经常不完整。比如有些项目只写了项目名和主管单位，没有明确写牲畜类型。这个时候脚本不应该停掉，而应该把能抓到的字段先入库，缺失字段留空，后续人工补充或二次清洗。

7.4 缺失字段怎么办？

本文采用三种策略：

第一，字段为空字符串，不抛异常。

第二，保留原始 HTML 到 data/raw/，方便后续排查。

第三，后续可以对空字段做二次补全，例如从正文、附件、标题关键词里推断。

举例：

项目名：草畜平衡示范项目
牲畜类型：空

这种不应该直接丢弃。因为项目名和地区可能已经有价值，后续可以用人工或规则补齐。

---

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

8.1 为什么先用 SQLite？

对于这类归档项目，SQLite 是很好的起点：

不用部署服务
单文件数据库
支持唯一约束
支持更新
方便导出 CSV
容易备份

如果你只是做日常项目台账，SQLite 足够。如果后面要多人协作、接口服务、后台管理系统，再迁移到 MySQL 或 PostgreSQL。

8.2 存储层代码：crawler/storage.py

# crawler/storage.py
from __future__ import annotations

import csv
import sqlite3
from pathlib import Path
from typing import Literal

from crawler.parser import RanchProject
from crawler.utils import ensure_dir, now_iso


SaveStatus = Literal["inserted", "updated", "skipped"]


class Storage:
    """
    SQLite 存储层。

    增量策略：
    - url 唯一
    - 新 URL：inserted
    - 老 URL 且 source_hash 变化：updated
    - 老 URL 且 source_hash 不变：skipped
    """

    def __init__(self, db_path: Path):
        self.db_path = db_path
        ensure_dir(db_path.parent)
        self.conn = sqlite3.connect(str(db_path))
        self.conn.row_factory = sqlite3.Row

    def init_db(self) -> None:
        self.conn.execute(
            """
            CREATE TABLE IF NOT EXISTS ranch_projects (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                project_name TEXT NOT NULL,
                region TEXT DEFAULT '',
                livestock_type TEXT DEFAULT '',
                cycle TEXT DEFAULT '',
                authority TEXT DEFAULT '',
                url TEXT NOT NULL UNIQUE,
                source_hash TEXT NOT NULL,
                first_seen_at TEXT NOT NULL,
                last_seen_at TEXT NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );
            """
        )

        self.conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_ranch_projects_region
            ON ranch_projects(region);
            """
        )

        self.conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_ranch_projects_livestock
            ON ranch_projects(livestock_type);
            """
        )

        self.conn.execute(
            """
            CREATE TABLE IF NOT EXISTS crawl_logs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT NOT NULL,
                status TEXT NOT NULL,
                message TEXT DEFAULT '',
                created_at TEXT NOT NULL
            );
            """
        )

        self.conn.commit()

    def save_project(self, project: RanchProject) -> SaveStatus:
        existing = self.conn.execute(
            """
            SELECT id, source_hash
            FROM ranch_projects
            WHERE url = ?
            """,
            (project.url,),
        ).fetchone()

        ts = now_iso()

        if existing is None:
            self.conn.execute(
                """
                INSERT INTO ranch_projects (
                    project_name,
                    region,
                    livestock_type,
                    cycle,
                    authority,
                    url,
                    source_hash,
                    first_seen_at,
                    last_seen_at,
                    created_at,
                    updated_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    project.project_name,
                    project.region,
                    project.livestock_type,
                    project.cycle,
                    project.authority,
                    project.url,
                    project.source_hash,
                    ts,
                    ts,
                    ts,
                    ts,
                ),
            )
            self.conn.commit()
            return "inserted"

        if existing["source_hash"] != project.source_hash:
            self.conn.execute(
                """
                UPDATE ranch_projects
                SET
                    project_name = ?,
                    region = ?,
                    livestock_type = ?,
                    cycle = ?,
                    authority = ?,
                    source_hash = ?,
                    last_seen_at = ?,
                    updated_at = ?
                WHERE url = ?
                """,
                (
                    project.project_name,
                    project.region,
                    project.livestock_type,
                    project.cycle,
                    project.authority,
                    project.source_hash,
                    ts,
                    ts,
                    project.url,
                ),
            )
            self.conn.commit()
            return "updated"

        self.conn.execute(
            """
            UPDATE ranch_projects
            SET last_seen_at = ?
            WHERE url = ?
            """,
            (ts, project.url),
        )
        self.conn.commit()
        return "skipped"

    def log(self, url: str, status: str, message: str = "") -> None:
        self.conn.execute(
            """
            INSERT INTO crawl_logs (url, status, message, created_at)
            VALUES (?, ?, ?, ?)
            """,
            (url, status, message, now_iso()),
        )
        self.conn.commit()

    def export_csv(self, csv_path: Path) -> None:
        ensure_dir(csv_path.parent)

        rows = self.conn.execute(
            """
            SELECT
                project_name,
                region,
                livestock_type,
                cycle,
                authority,
                url,
                first_seen_at,
                last_seen_at
            FROM ranch_projects
            ORDER BY id ASC
            """
        ).fetchall()

        with csv_path.open("w", encoding="utf-8-sig", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(
                [
                    "项目名",
                    "地区",
                    "牲畜类型",
                    "周期",
                    "主管单位",
                    "链接",
                    "首次发现时间",
                    "最近采集时间",
                ]
            )

            for row in rows:
                writer.writerow(
                    [
                        row["project_name"],
                        row["region"],
                        row["livestock_type"],
                        row["cycle"],
                        row["authority"],
                        row["url"],
                        row["first_seen_at"],
                        row["last_seen_at"],
                    ]
                )

    def count_projects(self) -> int:
        row = self.conn.execute(
            "SELECT COUNT(*) AS c FROM ranch_projects"
        ).fetchone()
        return int(row["c"])

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

8.3 字段映射表

中文字段	数据库字段	类型	示例值
项目名	project_name	TEXT	青岭牧区草畜平衡示范项目
地区	region	TEXT	青岭牧区
牲畜类型	livestock_type	TEXT	牦牛、藏羊
周期	cycle	TEXT	2025-2027
主管单位	authority	TEXT	青岭畜牧技术服务中心
链接	url	TEXT	http://127.0.0.1:8765/detail/1001.html
内容 hash	source_hash	TEXT	sha256 字符串
首次发现时间	first_seen_at	TEXT	2026-06-10T10:00:00+00:00
最近采集时间	last_seen_at	TEXT	2026-06-10T10:00:00+00:00

8.4 去重策略

本文采用 URL 唯一：

url TEXT NOT NULL UNIQUE

这个策略适合项目详情页比较稳定的站点。只要详情页链接不变，同一个项目就不会重复入库。

同时用 source_hash 判断内容是否变化：

if existing["source_hash"] != project.source_hash:
    update
else:
    skip

这就是增量归档的关键。

比如第一次采集：

1001.html → inserted
1002.html → inserted
1003.html → inserted

第二次采集：

1001.html 内容没变 → skipped
1002.html 内容修改 → updated
1003.html 内容没变 → skipped
1004.html 新项目 → inserted

这样数据库会越来越完整，而不是越跑越乱。

---

9️⃣ Pipeline 编排与本地 Demo

为了保证代码真实可运行，本文提供一个本地 demo 站点。它会生成几份 HTML 页面，然后用 Python 内置 HTTP Server 启动一个本地服务。

这样你不用依赖任何外部网站，也能完整跑通：

请求列表页 → 解析详情链接 → 请求详情页 → 解析字段 → 入库 → 导出 CSV

9.1 本地示例站点：examples/demo_site.py

# examples/demo_site.py
from __future__ import annotations

import threading
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path


DEMO_FILES = {
    "list.html": """
<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>牧区公开项目归档示例</title>
</head>
<body>
  <main>
    <h1>牧区公开项目列表</h1>
    <ul class="project-list">
      <li><a class="project-link" href="/detail/1001.html">青岭牧区草畜平衡示范项目</a></li>
      <li><a class="project-link" href="/detail/1002.html">河湾牧场肉牛标准化养殖提升项目</a></li>
      <li><a class="project-link" href="/detail/1003.html">北坡牧区羊群疫病防控服务项目</a></li>
      <li><a class="project-link" href="/detail/1004.html">沙丘牧场骆驼养殖基础设施改善项目</a></li>
    </ul>
  </main>
</body>
</html>
""",
    "detail/1001.html": """
<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>青岭牧区草畜平衡示范项目</title>
</head>
<body>
  <main>
    <h1>青岭牧区草畜平衡示范项目</h1>
    <table>
      <tr><th>地区</th><td>青岭牧区</td></tr>
      <tr><th>牲畜类型</th><td>牦牛、藏羊</td></tr>
      <tr><th>周期</th><td>2025-2027</td></tr>
      <tr><th>主管单位</th><td>青岭畜牧技术服务中心</td></tr>
    </table>
    <p>项目围绕草畜平衡、牧草补播、牲畜承载量监测等内容开展。</p>
  </main>
</body>
</html>
""",
    "detail/1002.html": """
<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>河湾牧场肉牛标准化养殖提升项目</title>
</head>
<body>
  <main>
    <h1>河湾牧场肉牛标准化养殖提升项目</h1>
    <dl>
      <dt>所属地区</dt><dd>河湾牧场片区</dd>
      <dt>畜种</dt><dd>肉牛</dd>
      <dt>建设周期</dt><dd>2024-2026</dd>
      <dt>组织单位</dt><dd>河湾农牧综合服务站</dd>
    </dl>
    <p>该项目重点改造牛舍通风、饮水和饲草储备设施。</p>
  </main>
</body>
</html>
""",
    "detail/1003.html": """
<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>北坡牧区羊群疫病防控服务项目</title>
</head>
<body>
  <main>
    <h1>北坡牧区羊群疫病防控服务项目</h1>
    <p>地区：北坡牧区</p>
    <p>主要牲畜：绵羊、山羊</p>
    <p>服务周期：2026年度</p>
    <p>实施单位：北坡动物疫病防控服务队</p>
    <p>项目内容包括免疫巡检、档案登记和养殖户技术指导。</p>
  </main>
</body>
</html>
""",
    "detail/1004.html": """
<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>沙丘牧场骆驼养殖基础设施改善项目</title>
</head>
<body>
  <main>
    <h1>沙丘牧场骆驼养殖基础设施改善项目</h1>
    <div class="meta">
      <div>牧区：沙丘牧场</div>
      <div>养殖类型：骆驼</div>
      <div>实施周期：2025年5月-2026年10月</div>
      <div>承担单位：沙丘牧场经营服务中心</div>
    </div>
    <p>项目用于改善棚圈、饮水点和饲草转运条件。</p>
  </main>
</body>
</html>
""",
}


def build_demo_site(root: Path) -> None:
    root.mkdir(parents=True, exist_ok=True)

    for relative_path, content in DEMO_FILES.items():
        target = root / relative_path
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(content.strip(), encoding="utf-8")


def serve_demo_site(root: Path, host: str = "127.0.0.1", port: int = 8765):
    """
    启动本地 HTTP 服务。

    返回 server 对象，调用 server.shutdown() 可停止。
    """

    class DemoHandler(SimpleHTTPRequestHandler):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, directory=str(root), **kwargs)

        def log_message(self, format, *args):
            # demo 时减少控制台噪音
            return

    server = ThreadingHTTPServer((host, port), DemoHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server

9.2 Pipeline：crawler/pipeline.py

# crawler/pipeline.py
from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

from crawler.config import SiteConfig
from crawler.fetcher import Fetcher, FetchError
from crawler.parser import Parser
from crawler.storage import Storage
from crawler.utils import ensure_dir, safe_filename_from_url


@dataclass
class CrawlStats:
    list_pages: int = 0
    detail_pages: int = 0
    inserted: int = 0
    updated: int = 0
    skipped: int = 0
    failed: int = 0


class RanchProjectCrawler:
    """
    牧场/牧区公开项目归档爬虫。
    """

    def __init__(self, config: SiteConfig):
        self.config = config
        self.fetcher = Fetcher(config)
        self.parser = Parser(config)
        self.storage = Storage(config.db_path)
        self.stats = CrawlStats()

    def run(self) -> CrawlStats:
        ensure_dir(self.config.data_dir)
        ensure_dir(self.config.raw_dir)

        self.storage.init_db()

        for list_path in self.config.list_paths:
            list_url = self.fetcher.build_url(list_path)
            self._crawl_list_page(list_url)

        self.storage.export_csv(self.config.csv_path)
        return self.stats

    def _crawl_list_page(self, list_url: str) -> None:
        try:
            result = self.fetcher.get(list_url)
            self.stats.list_pages += 1
            self._save_raw(result.url, result.text)

            detail_links = self.parser.parse_list_links(result.text, result.url)

            print(f"[LIST] {result.url} -> {len(detail_links)} detail links")

            for detail_url in detail_links:
                self.fetcher.polite_sleep()
                self._crawl_detail_page(detail_url, referer=result.url)

        except Exception as exc:
            self.stats.failed += 1
            self.storage.log(list_url, "failed", str(exc))
            print(f"[FAILED LIST] {list_url} -> {exc}")

    def _crawl_detail_page(self, detail_url: str, referer: str) -> None:
        try:
            result = self.fetcher.get(detail_url, referer=referer)
            self.stats.detail_pages += 1
            self._save_raw(result.url, result.text)

            project = self.parser.parse_detail(result.text, result.url)
            status = self.storage.save_project(project)
            self.storage.log(result.url, status, project.project_name)

            if status == "inserted":
                self.stats.inserted += 1
            elif status == "updated":
                self.stats.updated += 1
            else:
                self.stats.skipped += 1

            print(f"[{status.upper()}] {project.project_name} | {project.url}")

        except FetchError as exc:
            self.stats.failed += 1
            self.storage.log(detail_url, "fetch_failed", str(exc))
            print(f"[FETCH FAILED] {detail_url} -> {exc}")

        except Exception as exc:
            self.stats.failed += 1
            self.storage.log(detail_url, "parse_or_save_failed", str(exc))
            print(f"[FAILED DETAIL] {detail_url} -> {exc}")

    def _save_raw(self, url: str, html: str) -> None:
        filename = safe_filename_from_url(url)
        path = self.config.raw_dir / filename
        path.write_text(html, encoding="utf-8")

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

这个 Pipeline 做了几件事：

初始化数据库
请求列表页
保存列表页原始 HTML
解析详情页链接
逐个请求详情页
保存详情页原始 HTML
解析项目字段
入库
导出 CSV

我保留了比较朴素的 print() 输出。生产环境可以换成 logging，但文章里先让流程更直观。

9.3 命令行入口：main.py

# main.py
from __future__ import annotations

import argparse
from pathlib import Path

from crawler.config import SiteConfig
from crawler.pipeline import RanchProjectCrawler
from examples.demo_site import build_demo_site, serve_demo_site


def parse_args():
    parser = argparse.ArgumentParser(
        description="牧场/牧区公开项目归档采集器"
    )

    parser.add_argument(
        "--demo",
        action="store_true",
        help="启动本地 demo 站点并采集示例数据",
    )

    parser.add_argument(
        "--base-url",
        default=None,
        help="目标站点 base URL，例如 https://example.com",
    )

    parser.add_argument(
        "--list-path",
        action="append",
        default=None,
        help="列表页路径，可重复传入，例如 --list-path /projects/list.html",
    )

    parser.add_argument(
        "--db",
        default="data/ranch_projects.sqlite3",
        help="SQLite 数据库路径",
    )

    parser.add_argument(
        "--csv",
        default="data/ranch_projects.csv",
        help="CSV 导出路径",
    )

    parser.add_argument(
        "--respect-robots",
        action="store_true",
        help="请求前检查 robots.txt",
    )

    return parser.parse_args()


def build_config(args) -> SiteConfig:
    config = SiteConfig()

    if args.demo:
        demo_root = Path("demo_site")
        build_demo_site(demo_root)
        serve_demo_site(demo_root, host="127.0.0.1", port=8765)
        config.base_url = "http://127.0.0.1:8765"
        config.list_paths = ["/list.html"]

    if args.base_url:
        config.base_url = args.base_url

    if args.list_path:
        config.list_paths = args.list_path

    config.db_path = Path(args.db)
    config.csv_path = Path(args.csv)
    config.respect_robots = args.respect_robots

    return config


def main():
    args = parse_args()
    config = build_config(args)

    crawler = RanchProjectCrawler(config)

    try:
        stats = crawler.run()
        total = crawler.storage.count_projects()

        print("\n=== Crawl Finished ===")
        print(f"List pages   : {stats.list_pages}")
        print(f"Detail pages : {stats.detail_pages}")
        print(f"Inserted     : {stats.inserted}")
        print(f"Updated      : {stats.updated}")
        print(f"Skipped      : {stats.skipped}")
        print(f"Failed       : {stats.failed}")
        print(f"Total in DB  : {total}")
        print(f"DB path      : {config.db_path}")
        print(f"CSV path     : {config.csv_path}")

    finally:
        crawler.close()


if __name__ == "__main__":
    main()

9.4 __init__.py

两个空文件即可：

# crawler/__init__.py

# examples/__init__.py

---

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

9.1 本地 demo 运行

安装依赖后，执行：

python main.py --demo

你会看到类似输出：

[LIST] http://127.0.0.1:8765/list.html -> 4 detail links
[INSERTED] 青岭牧区草畜平衡示范项目 | http://127.0.0.1:8765/detail/1001.html
[INSERTED] 河湾牧场肉牛标准化养殖提升项目 | http://127.0.0.1:8765/detail/1002.html
[INSERTED] 北坡牧区羊群疫病防控服务项目 | http://127.0.0.1:8765/detail/1003.html
[INSERTED] 沙丘牧场骆驼养殖基础设施改善项目 | http://127.0.0.1:8765/detail/1004.html

=== Crawl Finished ===
List pages   : 1
Detail pages : 4
Inserted     : 4
Updated      : 0
Skipped      : 0
Failed       : 0
Total in DB  : 4
DB path      : data/ranch_projects.sqlite3
CSV path     : data/ranch_projects.csv

第二次再跑：

python main.py --demo

输出会变成：

[LIST] http://127.0.0.1:8765/list.html -> 4 detail links
[SKIPPED] 青岭牧区草畜平衡示范项目 | http://127.0.0.1:8765/detail/1001.html
[SKIPPED] 河湾牧场肉牛标准化养殖提升项目 | http://127.0.0.1:8765/detail/1002.html
[SKIPPED] 北坡牧区羊群疫病防控服务项目 | http://127.0.0.1:8765/detail/1003.html
[SKIPPED] 沙丘牧场骆驼养殖基础设施改善项目 | http://127.0.0.1:8765/detail/1004.html

=== Crawl Finished ===
List pages   : 1
Detail pages : 4
Inserted     : 0
Updated      : 0
Skipped      : 4
Failed       : 0
Total in DB  : 4
DB path      : data/ranch_projects.sqlite3
CSV path     : data/ranch_projects.csv

这就说明增量逻辑生效了。

9.2 输出在哪里？

数据库：

data/ranch_projects.sqlite3

CSV：

data/ranch_projects.csv

原始 HTML：

data/raw/

保留原始 HTML 是我比较坚持的习惯。因为解析错误时，只有结构化数据是不够的，你需要回头看当时页面到底长什么样。

9.3 示例结果

CSV 中会有类似数据：

项目名	地区	牲畜类型	周期	主管单位	链接
青岭牧区草畜平衡示范项目	青岭牧区	牦牛、藏羊	2025-2027	青岭畜牧技术服务中心	http://127.0.0.1:8765/detail/1001.html
河湾牧场肉牛标准化养殖提升项目	河湾牧场片区	肉牛	2024-2026	河湾农牧综合服务站	http://127.0.0.1:8765/detail/1002.html
北坡牧区羊群疫病防控服务项目	北坡牧区	绵羊、山羊	2026年度	北坡动物疫病防控服务队	http://127.0.0.1:8765/detail/1003.html
沙丘牧场骆驼养殖基础设施改善项目	沙丘牧场	骆驼	2025年5月-2026年10月	沙丘牧场经营服务中心	http://127.0.0.1:8765/detail/1004.html

9.4 换成真实站点怎么跑？

假设真实站点是：

https://example.com

列表页是：

https://example.com/ranch/projects/index.html

则可以运行：

python main.py \
  --base-url https://example.com \
  --list-path /ranch/projects/index.html \
  --respect-robots

如果列表页链接选择器不是 a.project-link，就改 crawler/config.py：

list_link_selector: str = "ul.news-list a"

如果详情页标题不是 h1，也改配置：

title_selector: str = ".article-title"

如果正文容器不是 main：

detail_container_selector: str = ".article-content"

换站点时，最常改的就是这三个选择器。

---

🔟 常见问题与排错

10.1 遇到 403 怎么办？

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

路径不允许访问
User-Agent 被拒绝
Referer 校验
IP 访问频率异常
页面需要登录

建议按顺序排查：

第一，确认页面是否公开可访问。浏览器无登录状态能不能打开？

第二，检查 robots.txt。如果目标路径明确禁止抓取，就不要抓。

第三，降低频率。把 sleep_min 和 sleep_max 调大：

sleep_min = 2.0
sleep_max = 5.0

第四，设置合理 User-Agent。不要用空 UA，也不要伪装得过分离谱。

第五，如果必须登录才能看，不建议继续绕。改用官方接口、授权数据或手动导出。

代理不是万能解。代理更适合解决网络出口问题，不应该用来规避访问限制。

10.2 遇到 429 怎么办？

429 通常表示请求太频繁。

处理思路：

降低请求频率
增加随机 sleep
减少并发
指数退避重试
分批运行

本文 Fetcher 已经对 429 做了重试和退避：

if response.status_code in {429, 500, 502, 503, 504}:
    raise FetchError(...)

但这不代表可以继续高频抓。退避只是补救，频控才是根本。

10.3 HTML 抓到空壳怎么办？

有些页面在浏览器里能看到内容，但 requests 抓下来只有：

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

这说明页面是动态渲染的。

排查方法：

第一，打开浏览器开发者工具 Network，看是否有 JSON 接口。

第二，如果有接口，优先抓接口。接口通常更稳定、更干净。

第三，如果没有明显接口，再考虑 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 解决时，不要急着上浏览器自动化。

10.4 解析报错怎么办？

解析报错通常是页面结构和选择器不匹配。

比如列表页没有抓到详情链接：

[LIST] xxx -> 0 detail links

这时先保存页面：

data/raw/

打开 raw HTML 看真实结构，再调整：

list_link_selector = "a.project-title"

如果详情字段为空，检查正文容器：

detail_container_selector = ".content"

很多时候不是代码坏了，而是选择器写错了。

10.5 编码或乱码怎么办？

乱码常见于老站点。本文里做了一个处理：

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

如果还是乱码，可以手动指定：

response.encoding = "utf-8"

或者：

response.encoding = "gb18030"

CSV 导出时使用：

encoding="utf-8-sig"

这样 Excel 打开中文时更友好。

10.6 字段抽不全怎么办？

字段抽不全很正常。建议分三步：

第一，先确认详情页是否真的有这个字段。

第二，增加字段别名：

"authority": [
    "主管单位",
    "发布单位",
    "组织单位",
    "实施单位",
    "承担单位",
    "管理单位",
    "责任单位",
]

第三，增加兜底正则。

比如页面写的是：

本项目由某某服务中心负责组织实施。

可以增加规则：

r"由([^。；;]+?)负责组织实施"

但正则越多，误伤越多。我的建议是：先结构化解析，再谨慎做正文规则。

---

1️⃣1️⃣ 进阶优化

11.1 并发优化

本文默认串行，因为项目归档任务更重视稳定性。

如果你确实需要提速，可以考虑线程池：

from concurrent.futures import ThreadPoolExecutor, as_completed

def crawl_detail(detail_url):
    result = fetcher.get(detail_url)
    project = parser.parse_detail(result.text, result.url)
    return project

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(crawl_detail, url) for url in detail_links]
    for future in as_completed(futures):
        project = future.result()
        storage.save_project(project)

并发数不要一上来就开很大。项目公开页面采集，我一般从 2 或 3 开始。如果 429 增多，就降回串行。

如果站点本来就慢，盲目并发只会让失败率更高。

11.2 asyncio / httpx

如果详情页很多，且站点能承受合理并发，可以用 httpx.AsyncClient。但异步代码会增加复杂度，也要处理限速、重试、异常收集。

一个简化思路：

import asyncio
import httpx


async def fetch_one(client: httpx.AsyncClient, url: str) -> str:
    response = await client.get(url, timeout=10)
    response.raise_for_status()
    return response.text


async def fetch_many(urls: list[str]) -> list[str]:
    limits = httpx.Limits(max_connections=5)
    async with httpx.AsyncClient(limits=limits) as client:
        tasks = [fetch_one(client, url) for url in urls]
        return await asyncio.gather(*tasks)

我不建议初版就上异步。解析、存储、容错还没稳定时，异步只会让排错更痛苦。

11.3 Scrapy 版本思路

当你要采很多列表页、分页、详情页，并且需要更完整的调度、去重、日志、下载中间件，Scrapy 更合适。

Scrapy 的结构大概是：

items.py 定义字段
spider.py 解析列表和详情
pipelines.py 入库
middlewares.py 处理请求头、代理、重试
settings.py 控制并发、延迟、日志

示例 Item：

import scrapy


class RanchProjectItem(scrapy.Item):
    project_name = scrapy.Field()
    region = scrapy.Field()
    livestock_type = scrapy.Field()
    cycle = scrapy.Field()
    authority = scrapy.Field()
    url = scrapy.Field()
    source_hash = scrapy.Field()

Scrapy 的优点是体系完整，缺点是上手比 requests 稍重。对于长期项目，我会在 requests 版跑通之后，再考虑迁移 Scrapy。

11.4 断点续跑

断点续跑有几种做法。

第一种，直接依赖数据库 URL 唯一。本文已经做了，重复 URL 会 skipped。

第二种，维护已抓集合：

seen_urls = set(
    row["url"]
    for row in conn.execute("SELECT url FROM ranch_projects")
)

第三种，记录分页游标：

last_page=12
last_crawl_at=2026-06-10T10:00:00Z

项目归档里，我更喜欢数据库唯一约束，因为它简单可靠，不容易因为状态文件丢失导致重复。

11.5 日志与监控

生产环境不要只靠 print()。可以用 logging：

import logging

logging.basicConfig(
    filename="data/crawler.log",
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

logging.info("crawl started")
logging.warning("empty field: livestock_type")
logging.error("fetch failed", exc_info=True)

建议记录这些指标：

指标	含义
请求总数	本次访问了多少页面
成功数	成功解析多少页面
失败数	请求或解析失败多少
新增数	inserted
更新数	updated
跳过数	skipped
空字段率	哪些字段经常缺失

字段缺失率很有价值。比如 牲畜类型 为空很多，说明页面可能不直接写这个字段，需要换解析策略。

11.6 定时任务

Linux 可以用 cron：

crontab -e

每天凌晨 2 点运行：

0 2 * * * cd /path/to/ranch_project_archive && .venv/bin/python main.py --base-url https://example.com --list-path /projects/index.html --respect-robots >> data/cron.log 2>&1

如果流程更复杂，可以用 Airflow、Prefect 或 Dagster。但普通归档任务，cron 已经够用。

11.7 数据质量校验

入库后建议做简单校验：

def validate_project(project):
    warnings = []

    if not project.project_name:
        warnings.append("missing project_name")

    if not project.url:
        warnings.append("missing url")

    if not project.region:
        warnings.append("missing region")

    return warnings

字段不完整不一定要丢弃，但要记录。后续你才能知道哪类页面质量差，哪类规则需要加强。

11.8 内容 hash 的粒度优化

本文直接对正文文本做 hash：

source_hash = sha256_text(fallback_text)

这很简单，但也有一个问题：如果页面底部日期、访问量、推荐链接变化，也可能导致 hash 变化。

更稳的做法是只对核心字段做 hash：

source_hash = sha256_text(
    "|".join(
        [
            project_name,
            region,
            livestock_type,
            cycle,
            authority,
        ]
    )
)

两种方案各有取舍：

hash 方式	优点	缺点
正文 hash	能发现页面任何变化	容易被无关内容影响
字段 hash	只关注结构化字段变化	发现不了正文补充说明变化

如果是项目台账，我通常选择字段 hash。如果是网页归档，我会保存正文 hash 和字段 hash 两份。

11.9 附件采集

很多项目页面会挂 PDF、Word 或 Excel 附件。初版可以先保存附件链接：

def parse_attachments(soup, page_url):
    attachments = []
    for a in soup.select("a[href]"):
        href = a.get("href", "")
        if href.lower().endswith((".pdf", ".doc", ".docx", ".xls", ".xlsx")):
            attachments.append(urljoin(page_url, href))
    return attachments

但附件解析是另一个坑。PDF 可能是扫描件，Excel 可能有合并单元格，Word 可能格式很乱。建议第二阶段再做。

11.10 项目字段标准化

牲畜类型建议做标准化。比如：

牦牛、藏羊
绵羊、山羊
肉牛
骆驼

可以拆成多值字段：

def split_livestock_types(text: str) -> list[str]:
    for sep in ["、", "，", ",", "/", "及", "和"]:
        text = text.replace(sep, "|")
    return [item.strip() for item in text.split("|") if item.strip()]

长期看，最好建一个字典表：

raw_value → standard_value
藏羊 → 羊
绵羊 → 羊
山羊 → 羊
肉牛 → 牛
牦牛 → 牛

这样后续分析会更干净。

---

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

本文完成了一套“牧场/牧区公开项目归档”的 Python 采集方案。

我们做了这些事：

第一，明确了目标字段：

项目名、地区、牲畜类型、周期、主管单位、链接

第二，设计了完整流程：

采集 → 解析 → 清洗 → 存储 → 导出

第三，写了可运行代码：

requests 请求
BeautifulSoup 解析
SQLite 增量入库
CSV 导出
本地 demo 页面

第四，加入了归档任务很重要的增量逻辑：

URL 唯一
source_hash 判断内容变化
inserted / updated / skipped 三种状态

第五，说明了合规边界：

看 robots.txt
控制频率
不绕过登录和付费限制
不采集敏感信息

如果下一步继续扩展，我建议按这个顺序来：

1. 先把真实站点的选择器配置好，稳定跑通 100 条以内的数据。
2. 再补字段别名和正文兜底规则，提高字段完整率。
3. 然后加入日志、失败重试报表和定时任务。
4. 数据量变大后，再考虑 Scrapy、Playwright 或分布式调度。
5. 如果有附件，再单独开一条附件下载和解析链路。

我写爬虫一直有个习惯：先让它慢慢地、干净地、可重复地跑起来。能归档、能增量、能排错，比一开始追求高并发更重要。尤其是公开项目数据，价值不在于一次抓多少，而在于长期积累之后，它能变成一份稳定、清楚、可追溯的资料库。

最后再强调一句：采集公开信息时，技术只是工具，边界感更重要。尊重站点规则，控制访问频率，不采敏感信息，不绕过限制。这样写出来的脚本，才适合长期使用。

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

---

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

---

✅ 免责声明

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

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

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