Python 爬虫实战：绿色产品认证目录分页采集，抓取产品名、品牌、认证类别、有效期与详情链接

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

全文目录：

• • 🌟 开篇语
  • 0️⃣ 前言（Preface）
  • 1️⃣ 摘要（Abstract）
  • 2️⃣ 背景与需求（Why）
  • • 2.1 为什么要抓取绿色产品认证目录
    • 2.2 目标站点类型
    • 2.3 目标字段清单
  • 3️⃣ 合规与注意事项（必写）
  • • 3.1 robots.txt 基本说明
    • 3.2 控制频率，不要攻击式并发
    • 3.3 不采集敏感信息
    • 3.4 不绕过付费、登录和验证码限制
    • 3.5 结果使用也要谨慎
  • 4️⃣ 技术选型与整体流程（What / How）
  • • 4.1 静态、动态还是 API
    • 4.2 整体流程
    • 4.3 为什么选择 requests + bs4 + lxml
  • 5️⃣ 环境准备与依赖安装（可复现）
  • • 5.1 Python 版本
    • 5.2 创建虚拟环境
    • 5.3 安装依赖
    • 5.4 推荐项目结构
  • 6️⃣ 核心实现：请求层（Fetcher）
  • • 6.1 配置文件 config.py
    • 6.2 日志模块 logger.py
    • 6.3 请求层 fetcher.py
    • 6.4 headers 说明
    • 6.5 timeout 说明
    • 6.6 session/cookie 说明
    • 6.7 失败处理：重试与退避
  • 7️⃣ 核心实现：解析层（Parser）
  • • 7.1 数据模型 models.py
    • 7.2 清洗工具 cleaner.py
    • 7.3 解析方式
    • 7.4 parser.py 完整实现
    • 7.5 列表页如何拿详情链接
    • 7.6 详情页如何抽字段
    • 7.7 缺失字段怎么办
  • 8️⃣ 数据存储与导出（Storage）
  • • 8.1 字段映射表
    • 8.2 去重策略
    • 8.3 storage.py 完整实现
    • 8.4 为什么 CSV 用 utf-8-sig
  • 9️⃣ 运行方式与结果展示（必写）
  • • 9.1 主入口 main.py
    • 9.2 准备本地 fixture，保证代码可复现
    • 9.3 启动方式
    • 9.4 输出位置
    • 9.5 示例结果
  • 🔟 常见问题与排错（强烈建议写）
  • • 10.1 403 Forbidden 怎么办
    • 10.2 429 Too Many Requests 怎么办
    • 10.3 HTML 抓到空壳怎么办
    • 10.4 JSON 接口模式示例
    • 10.5 解析报错怎么办
    • 10.6 编码乱码怎么办
    • 10.7 链接重复怎么办
    • 10.8 有效期解析不准怎么办
  • 1️⃣1️⃣ 进阶优化（可选但加分）
  • • 11.1 并发优化
    • 11.2 asyncio 版本适合吗
    • 11.3 断点续跑
    • 11.4 SQLite 存储版本
    • 11.5 日志与监控
    • 11.6 定时任务
    • 11.7 变更监控
    • 11.8 Playwright 兜底方案
  • 1️⃣2️⃣ 总结与延伸阅读
  • 🌟 文末
  • • ✅ 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集
    • ✅ 免责声明

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

这篇文章要做的事情很明确：用 Python 抓取一个公开的绿色产品认证目录分页页面，提取产品名、品牌、认证类别、有效期、详情链接等字段，最终导出为 CSV 和 JSONL 文件，方便后续做认证信息聚合、检索和数据分析。

读完本文，你可以获得：

本文的示例不讨论绕过验证码、登录限制、付费限制，也不会介绍任何攻击式并发方式。认证类数据通常具有公共信息属性，但这不代表可以无节制抓取。做爬虫，首先要能跑，其次要跑得稳，最后还要跑得体面。

1️⃣ 摘要（Abstract）

本文围绕“绿色产品认证目录分页”这一典型目录型站点，使用 requests + BeautifulSoup + lxml 实现一个可复现的 Python 爬虫，将产品名、品牌、认证类别、有效期、链接等字段结构化导出到本地文件。

读完本文，你将掌握：

这篇文章的重点不是“几行代码把网页扒下来”，而是把一个看起来简单、实际很容易翻车的目录采集任务做成一个稳一点的工程。绿色产品认证目录这类页面往往字段标准化程度较高，但页面结构未必稳定；有些列表页只展示摘要，有些详情页才有完整字段，有些页面还可能由接口异步加载。这些情况都会在实现里考虑进去。

2️⃣ 背景与需求（Why）

2.1 为什么要抓取绿色产品认证目录

绿色产品认证目录属于典型的“公开目录型信息”。它的价值不在于单条记录，而在于规模化整理之后可以服务很多事情，例如：

第一，做数据分析。

比如统计不同认证类别下的产品数量、品牌分布、有效期分布，观察哪些品类的绿色产品认证更集中，哪些品牌持有的有效认证更多。

第二，做信息聚合。

很多企业、采购人员、研究人员只想快速检索“某个品牌是否有绿色产品认证”“某个产品的认证有效期到什么时候”“某个类别下有哪些认证产品”。如果每次都手动翻目录，效率很低。

第三，做自动化监控。

认证信息不是永久不变的。有效期会变化，证书状态可能更新，目录也可能新增记录。把目录定期采集下来，就可以做增量比对，发现新增、过期、变更等情况。

第四，做内部数据底座。

对于企业内部系统来说，认证数据可以进入供应商管理、产品合规审查、采购辅助决策、数据看板等流程。爬虫只是第一步，后面还可以接数据库、搜索引擎、可视化报表。

2.2 目标站点类型

本文假设目标站点是一个公开访问的绿色产品认证目录页面，具备下面这些特征：

• 无需登录即可访问；
• 页面有分页；
• 列表页包含若干产品条目；
• 列表页可以拿到详情页链接；
• 详情页包含产品名、品牌、认证类别、有效期等字段；
• 页面没有强制验证码，或验证码不是采集路径的一部分；
• robots.txt 或网站说明没有明确禁止访问相关目录。

现实中，认证目录页面可能有三种形式：

本文主线采用第一种和第二种的兼容设计：优先用 requests 抓取 HTML；如果抓到的是空壳页面，再观察浏览器网络请求，切换到接口模式；如果必须渲染，则在进阶部分讨论 Playwright 的处理思路。

2.3 目标字段清单

本文目标字段如下：

工程里还会额外保留几个辅助字段：

虽然用户最终只关心五个字段，但我个人建议爬认证类目录时一定要保留认证编号。只靠产品名和品牌去重很容易误伤，因为同一个品牌可能有多个型号，同一个产品也可能经历换证、变更和续期。认证编号才是更稳的主键候选。

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

3.1 robots.txt 基本说明

开始写代码之前，先确认目标站点的 robots.txt。robots.txt 是网站给爬虫看的访问建议文件，通常位于网站根路径下，例如：

https://example.com/robots.txt

你需要关注两类信息：

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

含义大致是：

• User-agent：规则适用的爬虫；
• Disallow：不希望爬虫访问的路径；
• Allow：允许访问的路径。

robots.txt 不是权限系统，但它是爬虫礼仪的重要部分。如果目标目录路径被明确禁止抓取，就不应该继续自动化采集。可以选择联系站点方获取授权接口、使用站点提供的数据下载功能，或者改为人工检索。

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

认证目录通常不是高频变化的数据，没必要用很高的并发去压站点。

建议：

• 单线程或低并发起步；
• 每次请求之间随机 sleep；
• 对 429、503 等响应进行退避重试；
• 不要在短时间内重复抓同一批页面；
• 生产环境设置最大页数、最大请求量和超时。

爬虫不是压力测试工具。尤其是公共服务类站点，访问者很多，脚本要尽量温和。

3.3 不采集敏感信息

本文只采集公开目录字段，例如产品名、品牌、认证类别、有效期、详情链接等。不要采集个人隐私、账号信息、联系方式、登录态数据或后台数据。

如果详情页中出现不适合入库的字段，比如联系人手机号、私人邮箱、身份证号等，应当主动过滤。

3.4 不绕过付费、登录和验证码限制

如果站点需要登录，优先使用官方 API、授权账号或站点允许的导出功能。

如果页面出现验证码，不要写识别、绕过或模拟破解逻辑。验证码本质上是在表达“这里不希望自动化访问或需要人工确认”。这类场景正确的做法是降低自动化程度，走授权方式，或者只抓取无需验证码的公开页面。

3.5 结果使用也要谨慎

认证信息具有时效性。今天抓到的有效期、状态、类别，未来可能变化。导出的结果应该带上抓取时间，并在使用时说明数据来源和更新时间。不要把一次性抓取结果包装成永久准确的数据源。

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

4.1 静态、动态还是 API

做网页采集，我一般先用浏览器打开目标页面，再做三个判断：

第一，看“查看网页源代码”里有没有目标数据。

如果源代码里能看到产品名、品牌、有效期，说明页面大概率是静态 HTML，适合 requests + BeautifulSoup/lxml。

第二，看 Network 里有没有返回 JSON 的接口。

如果源代码里没有数据，但浏览器页面能显示数据，打开开发者工具的 Network 面板，刷新页面，重点看 XHR 或 Fetch 请求。很多目录页其实是前端请求接口后渲染表格。这种情况最好直接抓接口，而不是上 Playwright。

第三，看是否必须执行复杂 JavaScript。

如果接口参数加密、页面结构强依赖浏览器环境，或者列表渲染逻辑比较复杂，再考虑 Playwright。Playwright 能解决问题，但成本更高，速度更慢，也更容易被误用。

本文示例属于“认证型目录分页”场景，优先按静态 HTML + 可选 JSON 接口来设计。

4.2 整体流程

流程可以概括为：

采集 Fetch
↓
解析 Parse
↓
清洗 Clean
↓
去重 Dedupe
↓
存储 Store
↓
复核 Review

展开之后：

读取配置
↓
创建 Session 和请求头
↓
循环分页 URL
↓
请求列表页
↓
解析详情页链接
↓
请求详情页
↓
抽取产品名、品牌、认证类别、有效期、链接
↓
字段清洗与日期规范化
↓
按 cert_no / URL / hash 去重
↓
写入 CSV 和 JSONL
↓
输出日志和失败清单

4.3 为什么选择 requests + bs4 + lxml

这个组合比较朴素，但适合目录页：

• requests：负责 HTTP 请求，易于设置 headers、timeout、session；
• BeautifulSoup：容错性强，适合解析结构不太规整的 HTML；
• lxml：解析速度快，也支持更强的 XPath；
• csv/json：标准库即可导出基础结果；
• sqlite3：标准库即可做简单持久化和去重。

如果后续规模变大，可以迁移到 Scrapy。Scrapy 更适合大批量任务、请求队列、自动去重、失败重试、管道处理和分布式扩展。本文先用普通 Python 项目写清楚核心逻辑，这样更容易看懂，也更方便改造成 Scrapy 版本。

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

5.1 Python 版本

建议使用 Python 3.10 及以上版本。本文代码在 Python 3.10+ 风格下编写，使用了类型注解、dataclass 和 pathlib。

查看 Python 版本：

python –version

建议输出类似：

Python 3.10.13

或者：

Python 3.11.x

5.2 创建虚拟环境

mkdir green_cert_crawler
cd green_cert_crawler

python -m venv .venv

macOS / Linux 激活：

source .venv/bin/activate

Windows 激活：

.venv\\Scripts\\activate

5.3 安装依赖

创建 requirements.txt：

requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.2.2
python-dateutil==2.9.0.post0

安装：

pip install -r requirements.txt

5.4 推荐项目结构

green_cert_crawler/
├── requirements.txt
├── README.md
├── data/
│ ├── output/
│ │ ├── green_products.csv
│ │ └── green_products.jsonl
│ └── logs/
│ └── crawler.log
├── fixtures/
│ ├── list_page_1.html
│ ├── detail_10001.html
│ ├── detail_10002.html
│ └── detail_10003.html
└── green_cert_crawler/
├── __init__.py
├── config.py
├── models.py
├── fetcher.py
├── parser.py
├── storage.py
├── cleaner.py
├── logger.py
└── main.py

这里我保留了 fixtures/ 目录，原因很简单：爬虫解析层必须能单独测试。真实站点结构可能变化，但只要你保存几份列表页和详情页样本，就能在本地反复调解析逻辑，不必每次都请求远程站点。

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

请求层要解决四件事：

6.1 配置文件 config.py

先写配置：

# green_cert_crawler/config.py

from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent
DATA_DIR = BASE_DIR / "data"
OUTPUT_DIR = DATA_DIR / "output"
LOG_DIR = DATA_DIR / "logs"

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

CSV_PATH = OUTPUT_DIR / "green_products.csv"
JSONL_PATH = OUTPUT_DIR / "green_products.jsonl"
LOG_PATH = LOG_DIR / "crawler.log"

# 这里用占位模板。实际使用时替换成目标站点的公开分页 URL。
# 示例：
# LIST_URL_TEMPLATE = "https://example.com/green-products?page={page}"
LIST_URL_TEMPLATE = "https://example.com/green-products?page={page}"

START_PAGE = 1
END_PAGE = 5

REQUEST_TIMEOUT = 15

# 每次请求之间的随机等待范围，单位：秒
MIN_DELAY = 0.8
MAX_DELAY = 2.0

MAX_RETRIES = 3

DEFAULT_HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0 Safari/537.36"
),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.7",
"Connection": "keep-alive",
}

这里的 User-Agent 不是为了伪装成真人做违规访问，而是避免裸 python-requests 被很多站点直接拒绝。请求头应保持中性，不要伪造登录态，不要塞非法 cookie。

6.2 日志模块 logger.py

# green_cert_crawler/logger.py

import logging
from logging.handlers import RotatingFileHandler

from .config import LOG_PATH

def get_logger(name: str = "green_cert_crawler") –> logging.Logger:
logger = logging.getLogger(name)

if logger.handlers:
return logger

logger.setLevel(logging.INFO)

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

file_handler = RotatingFileHandler(
LOG_PATH,
maxBytes=2 * 1024 * 1024,
backupCount=3,
encoding="utf-8",
)
file_handler.setFormatter(formatter)
file_handler.setLevel(logging.INFO)

console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
console_handler.setLevel(logging.INFO)

logger.addHandler(file_handler)
logger.addHandler(console_handler)

return logger

我建议任何超过几十行的爬虫都加日志。不要只靠 print。当任务跑到第 30 页失败时，日志能告诉你失败在哪个 URL、状态码是什么、重试了几次、最终有没有写入文件。

6.3 请求层 fetcher.py

# green_cert_crawler/fetcher.py

import random
import time
from typing import Optional
from urllib.parse import urljoin

import requests
from requests import Response
from requests.exceptions import RequestException

from .config import (
DEFAULT_HEADERS,
REQUEST_TIMEOUT,
MIN_DELAY,
MAX_DELAY,
MAX_RETRIES,
)
from .logger import get_logger

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

class Fetcher:
def __init__(
self,
base_url: str = "",
headers: Optional[dict] = None,
timeout: int = REQUEST_TIMEOUT,
) –> None:
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.logger = get_logger(self.__class__.__name__)

self.session = requests.Session()
self.session.headers.update(DEFAULT_HEADERS)

if headers:
self.session.headers.update(headers)

def build_url(self, url: str) –> str:
if url.startswith("http://") or url.startswith("https://"):
return url
if not self.base_url:
return url
return urljoin(self.base_url + "/", url.lstrip("/"))

def polite_sleep(self) –> None:
delay = random.uniform(MIN_DELAY, MAX_DELAY)
time.sleep(delay)

def get(self, url: str, referer: Optional[str] = None) –> str:
final_url = self.build_url(url)

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

last_error: Optional[Exception] = None

for attempt in range(1, MAX_RETRIES + 1):
try:
self.polite_sleep()

self.logger.info(
"GET %s | attempt=%s/%s",
final_url,
attempt,
MAX_RETRIES,
)

response = self.session.get(
final_url,
headers=headers,
timeout=self.timeout,
)

if response.status_code == 429:
retry_after = response.headers.get("Retry-After")
wait_seconds = int(retry_after) if retry_after and retry_after.isdigit() else 10
self.logger.warning(
"Received 429 Too Many Requests. Sleep %s seconds.",
wait_seconds,
)
time.sleep(wait_seconds)
continue

if 500 <= response.status_code < 600:
self.logger.warning(
"Server error status=%s url=%s",
response.status_code,
final_url,
)
time.sleep(2 * attempt)
continue

response.raise_for_status()
self._fix_encoding(response)

return response.text

except RequestException as exc:
last_error = exc
wait = 2 * attempt
self.logger.warning(
"Request failed url=%s attempt=%s error=%s sleep=%s",
final_url,
attempt,
exc,
wait,
)
time.sleep(wait)

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

@staticmethod
def _fix_encoding(response: Response) –> None:
"""
requests 有时会猜错中文页面编码。
apparent_encoding 来自 chardet/charset_normalizer 的判断结果。
"""
if not response.encoding or response.encoding.lower() in {"iso-8859-1", "ascii"}:
response.encoding = response.apparent_encoding

6.4 headers 说明

请求头里最常用的是：

"User-Agent": "Mozilla/5.0 …"
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.7"
"Referer": "上一页 URL"

Referer 有时是必要的。很多详情页会检查来源页，虽然不一定严格阻断，但带上来源页更接近正常浏览路径。

不建议做的事情：

• 不要伪造登录后的 cookie；
• 不要硬塞别人账号的 token；
• 不要用高频代理池冲击站点；
• 不要绕过验证码。

6.5 timeout 说明

爬虫必须设置 timeout。没有 timeout 的请求，可能因为一个连接挂住，导致整个任务卡死。

本文设置：

REQUEST_TIMEOUT = 15

15 秒不是固定标准。内网页面可以更短，跨境或慢站点可以更长。重点是不能没有。

6.6 session/cookie 说明

requests.Session() 的好处：

• 复用 TCP 连接；
• 自动保存同一会话中的 cookie；
• 统一管理 headers；
• 代码更简洁。

对于公开目录页，session 通常够用。如果站点明确需要登录，本文不建议继续模拟登录抓取，除非你有明确授权，并且符合站点使用条款。

6.7 失败处理：重试与退避

失败处理不能一味重试。本文的策略是：

• 网络异常：最多重试 3 次；
• 服务器 5xx：等待后重试；
• 429：读取 Retry-After，没有则默认等待；
• 4xx：通常不重试，尤其 403/404；
• 每次请求前随机 sleep；
• 失败写日志，便于后续复盘。

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

解析层是整个爬虫最容易变化的地方。网站改一个 class 名，代码就可能抓不到数据。因此解析层要写得容错一点。

7.1 数据模型 models.py

# green_cert_crawler/models.py

from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class GreenProductRecord:
product_name: str
brand: str
cert_category: str
valid_until: str
link: str

cert_no: str = ""
source_page: int = 0
fetched_at: str = ""
row_hash: str = ""

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

def public_dict(self) –> dict:
"""
最终给业务方看的核心字段。
"""
return {
"产品名": self.product_name,
"品牌": self.brand,
"认证类别": self.cert_category,
"有效期": self.valid_until,
"链接": self.link,
}

7.2 清洗工具 cleaner.py

# green_cert_crawler/cleaner.py

import hashlib
import re
from datetime import datetime
from typing import Iterable, Optional

from dateutil import parser as date_parser

SPACE_RE = re.compile(r"\\s+")
DATE_RANGE_SPLIT_RE = re.compile(r"[至~—\\-–]+")

def clean_text(value: Optional[str]) –> str:
if value is None:
return ""
value = value.replace("\\xa0", " ")
value = SPACE_RE.sub(" ", value)
return value.strip(" ：:\\t\\r\\n")

def first_non_empty(values: Iterable[str]) –> str:
for value in values:
value = clean_text(value)
if value:
return value
return ""

def normalize_date(value: str) –> str:
"""
将常见中文日期转成 YYYY-MM-DD。
如果无法解析，保留清洗后的原始值。
"""
value = clean_text(value)
if not value:
return ""

value = (
value.replace("年", "-")
.replace("月", "-")
.replace("日", "")
.replace("/", "-")
.replace(".", "-")
)

# 如果是 2024-01-01 至 2027-01-01，优先保留截止日期
parts = [clean_text(p) for p in DATE_RANGE_SPLIT_RE.split(value) if clean_text(p)]
candidate = parts[–1] if parts else value

try:
dt = date_parser.parse(candidate, fuzzy=True)
return dt.strftime("%Y-%m-%d")
except Exception:
return value

def make_hash(*parts: str) –> str:
raw = "||".join(clean_text(p) for p in parts)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()

日期清洗是认证类目录里很重要的一步。不同站点可能写：

2027-06-30
2027年6月30日
2024-06-30 至 2027-06-29
有效期至：2027/06/30

业务通常更关心“截止日期”，所以本文把日期区间最后一个日期作为 valid_until。

7.3 解析方式

本文采用 BeautifulSoup 的 CSS 选择器和文本键值匹配结合的方式。

列表页做两件事：

详情页做三件事：

7.4 parser.py 完整实现

# green_cert_crawler/parser.py

from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Dict, Iterable, List, Optional
from urllib.parse import urljoin

from bs4 import BeautifulSoup, Tag

from .cleaner import clean_text, first_non_empty, normalize_date, make_hash
from .models import GreenProductRecord

FIELD_ALIASES = {
"product_name": [
"产品名",
"产品名称",
"获证产品",
"认证产品",
"商品名称",
"名称",
],
"brand": [
"品牌",
"商标",
"产品品牌",
"注册商标",
],
"cert_category": [
"认证类别",
"产品类别",
"认证目录",
"类别",
"认证类型",
],
"valid_until": [
"有效期",
"证书有效期",
"有效期至",
"截止日期",
"有效截止日期",
],
"cert_no": [
"证书编号",
"认证编号",
"编号",
"证书号",
],
}

@dataclass
class ListItem:
detail_url: str
source_page: int
list_text: str = ""

class GreenProductParser:
def __init__(self, base_url: str = "") –> None:
self.base_url = base_url.rstrip("/")

def parse_list_page(
self,
html: str,
page_url: str,
source_page: int,
) –> List[ListItem]:
"""
从列表页提取详情链接。
这里写了多套兜底逻辑：
1. 优先找 table tbody tr a；
2. 再找包含 detail/view/info/cert 等关键词的 a；
3. 最后返回去重后的链接。
"""
soup = BeautifulSoup(html, "lxml")
items: List[ListItem] = []
seen = set()

candidate_links: List[Tag] = []

table_links = soup.select("table tbody tr a[href], table tr a[href]")
candidate_links.extend(table_links)

if not candidate_links:
candidate_links.extend(
soup.select(
"a[href*='detail'], "
"a[href*='view'], "
"a[href*='info'], "
"a[href*='cert'], "
"a[href*='product']"
)
)

for a in candidate_links:
href = clean_text(a.get("href"))
if not href:
continue

url = self.absolute_url(href, page_url)

if url in seen:
continue

seen.add(url)

parent_text = clean_text(a.find_parent("tr").get_text(" ", strip=True)) if a.find_parent("tr") else clean_text(a.get_text(" ", strip=True))

items.append(
ListItem(
detail_url=url,
source_page=source_page,
list_text=parent_text,
)
)

return items

def parse_detail_page(
self,
html: str,
detail_url: str,
source_page: int,
fetched_at: str,
) –> GreenProductRecord:
soup = BeautifulSoup(html, "lxml")

kv = {}
kv.update(self._extract_table_kv(soup))
kv.update(self._extract_dl_kv(soup))
kv.update(self._extract_colon_kv(soup))

product_name = self._pick(kv, FIELD_ALIASES["product_name"])
brand = self._pick(kv, FIELD_ALIASES["brand"])
cert_category = self._pick(kv, FIELD_ALIASES["cert_category"])
valid_until = normalize_date(self._pick(kv, FIELD_ALIASES["valid_until"]))
cert_no = self._pick(kv, FIELD_ALIASES["cert_no"])

if not product_name:
product_name = self._fallback_title(soup)

row_hash = make_hash(
cert_no,
product_name,
brand,
cert_category,
valid_until,
detail_url,
)

return GreenProductRecord(
product_name=product_name,
brand=brand,
cert_category=cert_category,
valid_until=valid_until,
link=detail_url,
cert_no=cert_no,
source_page=source_page,
fetched_at=fetched_at,
row_hash=row_hash,
)

def absolute_url(self, href: str, current_url: str = "") –> str:
href = clean_text(href)
if href.startswith("http://") or href.startswith("https://"):
return href

base = current_url or self.base_url
if base:
return urljoin(base, href)

return href

@staticmethod
def _pick(kv: Dict[str, str], aliases: Iterable[str]) –> str:
normalized = {clean_text(k): clean_text(v) for k, v in kv.items()}

for alias in aliases:
alias_clean = clean_text(alias)
if alias_clean in normalized and normalized[alias_clean]:
return normalized[alias_clean]

# 模糊匹配：例如 “产品名称：” “产品名称（中文）”
for key, value in normalized.items():
for alias in aliases:
if alias in key and value:
return value

return ""

@staticmethod
def _extract_table_kv(soup: BeautifulSoup) –> Dict[str, str]:
result: Dict[str, str] = {}

for tr in soup.select("tr"):
cells = tr.find_all(["th", "td"])
texts = [clean_text(c.get_text(" ", strip=True)) for c in cells]
texts = [t for t in texts if t]

if len(texts) >= 2:
# 常见结构：<tr><th>产品名称</th><td>xxx</td></tr>
key = texts[0]
value = texts[1]
if key and value and len(key) <= 30:
result[key] = value

# 另一种结构：一行里有多组 key/value
if len(texts) >= 4:
for i in range(0, len(texts) – 1, 2):
k = texts[i]
v = texts[i + 1]
if k and v and len(k) <= 30:
result[k] = v

return result

@staticmethod
def _extract_dl_kv(soup: BeautifulSoup) –> Dict[str, str]:
result: Dict[str, str] = {}

for dl in soup.select("dl"):
dts = dl.find_all("dt")
dds = dl.find_all("dd")

for dt, dd in zip(dts, dds):
key = clean_text(dt.get_text(" ", strip=True))
value = clean_text(dd.get_text(" ", strip=True))
if key and value:
result[key] = value

return result

@staticmethod
def _extract_colon_kv(soup: BeautifulSoup) –> Dict[str, str]:
"""
兜底处理：
有些页面不是表格，而是很多 div/p/span：
产品名称：水性内墙涂料
品牌：GreenHome
"""
result: Dict[str, str] = {}

text = soup.get_text("\\n", strip=True)
lines = [clean_text(line) for line in text.splitlines()]
lines = [line for line in lines if line]

pattern = re.compile(r"^(.{2,30}?)[：:]\\s*(.+)$")

for line in lines:
match = pattern.match(line)
if not match:
continue

key = clean_text(match.group(1))
value = clean_text(match.group(2))

if key and value:
result[key] = value

return result

@staticmethod
def _fallback_title(soup: BeautifulSoup) –> str:
candidates = [
soup.select_one("h1"),
soup.select_one("h2"),
soup.select_one(".title"),
soup.select_one(".detail-title"),
soup.title,
]

return first_non_empty(
c.get_text(" ", strip=True) if c else ""
for c in candidates
)

7.5 列表页如何拿详情链接

典型列表页结构如下：

<table>
<tbody>
<tr>
<td>1</td>
<td>水性内墙涂料</td>
<td>GreenHome</td>
<td><a href="/green-products/detail/10001">查看详情</a></td>
</tr>
</tbody>
</table>

可以用：

soup.select("table tbody tr a[href]")

但真实页面很少这么规矩。有些链接文字叫“详情”，有些叫“查看”，有些整行可点击，有些 href 里包含 info、view 或 cert。所以 parser 里写了多种兜底选择器。

7.6 详情页如何抽字段

详情页可能是这种结构：

<table>
<tr>
<th>产品名称</th>
<td>水性内墙涂料</td>
</tr>
<tr>
<th>品牌</th>
<td>GreenHome</td>
</tr>
<tr>
<th>认证类别</th>
<td>绿色产品认证 / 涂料</td>
</tr>
<tr>
<th>有效期至</th>
<td>2027年6月30日</td>
</tr>
</table>

也可能是：

<dl>
<dt>产品名称</dt>
<dd>节水型坐便器</dd>
<dt>品牌</dt>
<dd>CleanBath</dd>
</dl>

还可能是：

<div>产品名称：环保型人造板</div>
<div>品牌：ForestBoard</div>
<div>有效期：2024-01-01 至 2027-01-01</div>

因此解析层不要只写死一种结构。本文实现了：

• table 键值抽取；
• dl 键值抽取；
• 冒号文本抽取；
• title 兜底；
• 字段别名匹配。

7.7 缺失字段怎么办

我的建议是：缺失字段不要让程序崩溃，应当保留空字符串，并写入日志或失败清单。

例如：

brand = self._pick(kv, FIELD_ALIASES["brand"])

如果品牌缺失，返回 ""。后续入库时仍然保留该记录，因为产品名、认证类别和链接可能仍然有价值。

但如果产品名和链接都缺失，那就应该丢弃。因为这类记录没有基本识别意义。

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

本文选择 CSV + JSONL 起步。

原因：

• CSV 适合 Excel、WPS、数据分析人员直接打开；
• JSONL 适合程序继续处理；
• 两者都轻量，不需要额外部署数据库；
• 后续迁移 SQLite/MySQL 很方便。

8.1 字段映射表

8.2 去重策略

认证类目录适合“编号优先去重”。

顺序建议：

对应代码：

def dedupe_key(record: GreenProductRecord) –> str:
if record.cert_no:
return f"cert_no:{record.cert_no}"
if record.link:
return f"url:{record.link}"
return f"hash:{record.row_hash}"

8.3 storage.py 完整实现

# green_cert_crawler/storage.py

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

from .models import GreenProductRecord

CSV_FIELDS = [
"产品名",
"品牌",
"认证类别",
"有效期",
"链接",
"认证编号",
"来源页码",
"抓取时间",
"内容哈希",
]

def dedupe_key(record: GreenProductRecord) –> str:
if record.cert_no:
return f"cert_no:{record.cert_no}"
if record.link:
return f"url:{record.link}"
return f"hash:{record.row_hash}"

class RecordStore:
def __init__(self) –> None:
self._seen: Set[str] = set()
self._records: List[GreenProductRecord] = []

def add(self, record: GreenProductRecord) –> bool:
key = dedupe_key(record)

if key in self._seen:
return False

# 产品名和链接都没有，基本不可用
if not record.product_name and not record.link:
return False

self._seen.add(key)
self._records.append(record)
return True

def extend(self, records: Iterable[GreenProductRecord]) –> int:
count = 0
for record in records:
if self.add(record):
count += 1
return count

@property
def records(self) –> List[GreenProductRecord]:
return self._records

def save_csv(self, path: Path) –> None:
path.parent.mkdir(parents=True, exist_ok=True)

with path.open("w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
writer.writeheader()

for record in self._records:
writer.writerow(
{
"产品名": record.product_name,
"品牌": record.brand,
"认证类别": record.cert_category,
"有效期": record.valid_until,
"链接": record.link,
"认证编号": record.cert_no,
"来源页码": record.source_page,
"抓取时间": record.fetched_at,
"内容哈希": record.row_hash,
}
)

def save_jsonl(self, path: Path) –> None:
path.parent.mkdir(parents=True, exist_ok=True)

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

8.4 为什么 CSV 用 utf-8-sig

很多中文用户会用 Excel 直接打开 CSV。如果用普通 utf-8，有些 Windows 环境下可能出现乱码。utf-8-sig 会在文件头写入 BOM，Excel 更容易识别。

这不是最优雅的编码方案，但很实用。

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

9.1 主入口 main.py

# green_cert_crawler/main.py

from __future__ import annotations

import argparse
from datetime import datetime
from pathlib import Path
from typing import List

from .config import (
CSV_PATH,
JSONL_PATH,
LIST_URL_TEMPLATE,
START_PAGE,
END_PAGE,
)
from .fetcher import Fetcher, FetchError
from .logger import get_logger
from .models import GreenProductRecord
from .parser import GreenProductParser
from .storage import RecordStore

def now_str() –> str:
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def crawl_pages(
list_url_template: str,
start_page: int,
end_page: int,
base_url: str = "",
) –> List[GreenProductRecord]:
logger = get_logger("main")

fetcher = Fetcher(base_url=base_url)
parser = GreenProductParser(base_url=base_url)

all_records: List[GreenProductRecord] = []

for page in range(start_page, end_page + 1):
list_url = list_url_template.format(page=page)
logger.info("Start list page page=%s url=%s", page, list_url)

try:
list_html = fetcher.get(list_url)
except FetchError as exc:
logger.error("Failed to fetch list page=%s error=%s", page, exc)
continue

list_items = parser.parse_list_page(
html=list_html,
page_url=list_url,
source_page=page,
)

logger.info("List page=%s found detail links=%s", page, len(list_items))

if not list_items:
logger.warning("No detail links found on page=%s", page)
continue

for item in list_items:
try:
detail_html = fetcher.get(item.detail_url, referer=list_url)
record = parser.parse_detail_page(
html=detail_html,
detail_url=item.detail_url,
source_page=item.source_page,
fetched_at=now_str(),
)
all_records.append(record)

logger.info(
"Parsed record product=%s brand=%s cert_no=%s",
record.product_name,
record.brand,
record.cert_no,
)

except Exception as exc:
logger.exception(
"Failed to parse detail url=%s error=%s",
item.detail_url,
exc,
)

return all_records

def crawl_from_fixtures(fixtures_dir: Path) –> List[GreenProductRecord]:
"""
本地样例模式：用于验证解析逻辑是否可运行。
不访问远程站点。
"""
logger = get_logger("fixtures")
parser = GreenProductParser(base_url="https://example.com")

list_page = fixtures_dir / "list_page_1.html"

if not list_page.exists():
raise FileNotFoundError(f"Fixture not found: {list_page}")

list_html = list_page.read_text(encoding="utf-8")
list_items = parser.parse_list_page(
html=list_html,
page_url="https://example.com/green-products?page=1",
source_page=1,
)

logger.info("Fixture list found detail links=%s", len(list_items))

records: List[GreenProductRecord] = []

for item in list_items:
detail_name = item.detail_url.rstrip("/").split("/")[–1]
detail_file = fixtures_dir / f"{detail_name}.html"

if not detail_file.exists():
logger.warning("Fixture detail file missing: %s", detail_file)
continue

detail_html = detail_file.read_text(encoding="utf-8")
record = parser.parse_detail_page(
html=detail_html,
detail_url=item.detail_url,
source_page=item.source_page,
fetched_at=now_str(),
)
records.append(record)

return records

def main() –> None:
arg_parser = argparse.ArgumentParser(
description="Green product certification directory crawler"
)
arg_parser.add_argument(
"–start-page",
type=int,
default=START_PAGE,
help="Start page number",
)
arg_parser.add_argument(
"–end-page",
type=int,
default=END_PAGE,
help="End page number",
)
arg_parser.add_argument(
"–list-url-template",
type=str,
default=LIST_URL_TEMPLATE,
help="List URL template, use {page} as page placeholder",
)
arg_parser.add_argument(
"–base-url",
type=str,
default="",
help="Base URL for relative links",
)
arg_parser.add_argument(
"–fixtures",
type=str,
default="",
help="Read local fixture HTML files instead of requesting remote pages",
)

args = arg_parser.parse_args()

logger = get_logger("main")
store = RecordStore()

if args.fixtures:
records = crawl_from_fixtures(Path(args.fixtures))
else:
records = crawl_pages(
list_url_template=args.list_url_template,
start_page=args.start_page,
end_page=args.end_page,
base_url=args.base_url,
)

added = store.extend(records)

store.save_csv(CSV_PATH)
store.save_jsonl(JSONL_PATH)

logger.info("Done. parsed=%s saved=%s", len(records), added)
logger.info("CSV saved to %s", CSV_PATH)
logger.info("JSONL saved to %s", JSONL_PATH)

print(f"Done. parsed={len(records)}, saved={added}")
print(f"CSV: {CSV_PATH}")
print(f"JSONL: {JSONL_PATH}")

if __name__ == "__main__":
main()

9.2 准备本地 fixture，保证代码可复现

创建 fixtures/list_page_1.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>绿色产品认证目录</title>
</head>
<body>
<h1>绿色产品认证目录</h1>
<table>
<thead>
<tr>
<th>序号</th>
<th>产品名</th>
<th>品牌</th>
<th>认证类别</th>
<th>详情</th>
</tr>
</thead>
<tbody>
<tr>
<td>1</td>
<td>水性内墙涂料</td>
<td>GreenHome</td>
<td>绿色产品认证 / 涂料</td>
<td><a href="/green-products/detail_10001">查看详情</a></td>
</tr>
<tr>
<td>2</td>
<td>节水型坐便器</td>
<td>CleanBath</td>
<td>绿色产品认证 / 卫生陶瓷</td>
<td><a href="/green-products/detail_10002">查看详情</a></td>
</tr>
<tr>
<td>3</td>
<td>环保型人造板</td>
<td>ForestBoard</td>
<td>绿色产品认证 / 人造板和木质地板</td>
<td><a href="/green-products/detail_10003">查看详情</a></td>
</tr>
</tbody>
</table>
</body>
</html>

创建 fixtures/detail_10001.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>水性内墙涂料</title>
</head>
<body>
<h1>水性内墙涂料</h1>
<table>
<tr><th>证书编号</th><td>GPC-2024-000001</td></tr>
<tr><th>产品名称</th><td>水性内墙涂料</td></tr>
<tr><th>品牌</th><td>GreenHome</td></tr>
<tr><th>认证类别</th><td>绿色产品认证 / 涂料</td></tr>
<tr><th>有效期至</th><td>2027年6月30日</td></tr>
</table>
</body>
</html>

创建 fixtures/detail_10002.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>节水型坐便器</title>
</head>
<body>
<h1>节水型坐便器</h1>
<dl>
<dt>认证编号</dt>
<dd>GPC-2024-000002</dd>
<dt>产品名</dt>
<dd>节水型坐便器</dd>
<dt>商标</dt>
<dd>CleanBath</dd>
<dt>产品类别</dt>
<dd>绿色产品认证 / 卫生陶瓷</dd>
<dt>证书有效期</dt>
<dd>2024-07-01 至 2027-06-30</dd>
</dl>
</body>
</html>

创建 fixtures/detail_10003.html：

<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>环保型人造板</title>
</head>
<body>
<h1>环保型人造板</h1>
<div class="detail">
<p>证书号：GPC-2024-000003</p>
<p>产品名称：环保型人造板</p>
<p>产品品牌：ForestBoard</p>
<p>认证目录：绿色产品认证 / 人造板和木质地板</p>
<p>有效截止日期：2027/12/31</p>
</div>
</body>
</html>

9.3 启动方式

本地 fixture 测试：

python -m green_cert_crawler.main –fixtures fixtures

真实分页目录抓取：

python -m green_cert_crawler.main \\
–base-url "https://example.com" \\
–list-url-template "https://example.com/green-products?page={page}" \\
–start-page 1 \\
–end-page 10

注意：真实站点使用前要先检查 robots.txt、页面访问规则和目标路径是否允许自动化访问。

9.4 输出位置

运行完成后输出：

data/output/green_products.csv
data/output/green_products.jsonl
data/logs/crawler.log

9.5 示例结果

CSV 中大致会得到：

JSONL 中每行是一条记录：

{"product_name":"水性内墙涂料","brand":"GreenHome","cert_category":"绿色产品认证 / 涂料","valid_until":"2027-06-30","link":"https://example.com/green-products/detail_10001","cert_no":"GPC-2024-000001","source_page":1,"fetched_at":"2026-06-09 10:30:21","row_hash":"…"}

我个人更喜欢同时保留 CSV 和 JSONL。CSV 给人看，JSONL 给程序看。很多时候数据工程做得顺不顺，不取决于你爬得多快，而取决于你后面能不能方便地复查、回放和修复。

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

10.1 403 Forbidden 怎么办

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

排查顺序：

第一步，降低频率。

把请求间隔调大，例如：

MIN_DELAY = 2.0
MAX_DELAY = 5.0

第二步，确认 headers。

至少设置：

User–Agent
Accept
Accept–Language
Referer

第三步，检查 robots.txt 和站点说明。

如果目标路径明确不允许抓取，不要继续。

第四步，判断是否需要授权。

如果页面必须登录，应该走授权接口或人工导出，不要模拟绕过。

10.2 429 Too Many Requests 怎么办

429 表示请求太频繁。

处理方式：

• 降低速度；
• 增加随机 sleep；
• 读取 Retry-After；
• 减少页数；
• 改成增量抓取；
• 避免重复请求同一详情页。

代码里已经做了基础处理：

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

不要看到 429 就上代理池猛冲。那不是优化，是把问题扩大。

10.3 HTML 抓到空壳怎么办

如果 requests.get(url).text 里只有：

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

说明数据可能由 JavaScript 渲染。

排查方法：

如果接口是公开的、无需绕过限制，可以改用接口抓取。

示例接口返回可能是：

{
"code": 200,
"data": {
"records": [
{
"productName": "水性内墙涂料",
"brand": "GreenHome",
"certCategory": "绿色产品认证 / 涂料",
"validUntil": "2027-06-30",
"detailUrl": "/green-products/detail_10001",
"certNo": "GPC-2024-000001"
}
],
"page": 1,
"totalPage": 10
}
}

这种情况下可以增加 JSON 解析模块。

10.4 JSON 接口模式示例

# green_cert_crawler/json_parser.py

from typing import List
from urllib.parse import urljoin

from .cleaner import clean_text, normalize_date, make_hash
from .models import GreenProductRecord

def parse_json_records(
payload: dict,
base_url: str,
source_page: int,
fetched_at: str,
) –> List[GreenProductRecord]:
data = payload.get("data") or {}
records = data.get("records") or data.get("list") or []

result: List[GreenProductRecord] = []

for item in records:
product_name = clean_text(
item.get("productName")
or item.get("product_name")
or item.get("产品名称")
)
brand = clean_text(
item.get("brand")
or item.get("brandName")
or item.get("品牌")
)
cert_category = clean_text(
item.get("certCategory")
or item.get("category")
or item.get("认证类别")
)
valid_until = normalize_date(
item.get("validUntil")
or item.get("expireDate")
or item.get("有效期")
or ""
)
cert_no = clean_text(
item.get("certNo")
or item.get("cert_no")
or item.get("证书编号")
)
link = clean_text(
item.get("detailUrl")
or item.get("url")
or item.get("链接")
)

if link and not link.startswith("http"):
link = urljoin(base_url.rstrip("/") + "/", link.lstrip("/"))

row_hash = make_hash(
cert_no,
product_name,
brand,
cert_category,
valid_until,
link,
)

result.append(
GreenProductRecord(
product_name=product_name,
brand=brand,
cert_category=cert_category,
valid_until=valid_until,
link=link,
cert_no=cert_no,
source_page=source_page,
fetched_at=fetched_at,
row_hash=row_hash,
)
)

return result

请求 JSON 接口时要注意：

response = session.get(api_url, params={"page": page, "size": 20}, timeout=15)
payload = response.json()

不要把接口参数写死到字符串里，尽量用 params，这样编码更安全，也更方便调试。

10.5 解析报错怎么办

常见报错包括：

AttributeError: 'NoneType' object has no attribute 'get_text'
IndexError: list index out of range
KeyError: 'data'

解决思路：

第一，所有选择器都要判断空。

不要写：

title = soup.select_one("h1").get_text(strip=True)

应该写：

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

第二，列表索引要谨慎。

不要假设每行都有 5 个 td：

tds = tr.find_all("td")
if len(tds) >= 5:
...

第三，保存失败页面。

解析失败时，把 HTML 保存到本地：

from pathlib import Path

debug_dir = Path("data/debug")
debug_dir.mkdir(parents=True, exist_ok=True)

(debug_dir / "failed_detail.html").write_text(html, encoding="utf-8")

有了失败样本，修选择器会快很多。

10.6 编码乱码怎么办

中文网页常见编码：

• UTF-8；
• GBK；
• GB2312；
• GB18030。

requests 有时猜错编码，导致中文乱码。本文在 fetcher 里用了：

if not response.encoding or response.encoding.lower() in {"iso-8859-1", "ascii"}:
response.encoding = response.apparent_encoding

如果仍然乱码，可以手动指定：

response.encoding = "gb18030"

导出 CSV 时建议：

encoding="utf-8-sig"

这样 Excel 打开更稳。

10.7 链接重复怎么办

列表页可能出现重复链接，例如标题链接和“查看详情”链接同时指向同一个详情页。

解决方法：

seen = set()

for url in urls:
if url in seen:
continue
seen.add(url)

入库层再做一次去重，不要只依赖解析层。

10.8 有效期解析不准怎么办

日期解析是半结构化文本里最烦人的部分。比如：

有效期：长期
有效期至：—
有效期：2024年1月1日至2027年1月1日
证书有效期：自发证之日起三年

对于无法解析成日期的值，不要强行转。保留原始值更安全。

本文的 normalize_date 逻辑是：

• 能解析就转成 YYYY-MM-DD；
• 日期区间取最后一个日期；
• 解析失败就返回原始清洗值。

这比直接抛异常更适合采集任务。

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

11.1 并发优化

如果目标站点允许合理并发，可以用线程池提高速度。认证目录通常详情页多，I/O 等待明显，线程池比较合适。

示例：

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_one_detail(fetcher, parser, item, list_url):
html = fetcher.get(item.detail_url, referer=list_url)
return parser.parse_detail_page(
html=html,
detail_url=item.detail_url,
source_page=item.source_page,
fetched_at=now_str(),
)

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

for future in as_completed(futures):
try:
record = future.result()
store.add(record)
except Exception as exc:
logger.exception("Detail task failed: %s", exc)

我建议从 max_workers=2 或 3 起步，不要上来就几十上百。目录站点不是越快越好，稳定和合规更重要。

11.2 asyncio 版本适合吗

可以，但不一定必要。

如果你要抓几万页，aiohttp 或 Scrapy 会更合适。但如果只是几百页目录，requests + 轻微并发 已经足够。

技术选型不要为了炫技。能稳定运行、便于维护，才是更好的选择。

11.3 断点续跑

断点续跑解决的问题是：爬到第 8 页断了，下次不用从第 1 页重来。

简单方案：保存已抓链接。

# green_cert_crawler/checkpoint.py

from pathlib import Path
from typing import Set

class Checkpoint:
def __init__(self, path: Path) –> None:
self.path = path
self.path.parent.mkdir(parents=True, exist_ok=True)
self.seen: Set[str] = set()
self.load()

def load(self) –> None:
if not self.path.exists():
return

with self.path.open("r", encoding="utf-8") as f:
for line in f:
value = line.strip()
if value:
self.seen.add(value)

def contains(self, key: str) –> bool:
return key in self.seen

def add(self, key: str) –> None:
if key in self.seen:
return

self.seen.add(key)

with self.path.open("a", encoding="utf-8") as f:
f.write(key + "\\n")

使用：

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

if checkpoint.contains(item.detail_url):
continue

# 抓取成功后
checkpoint.add(item.detail_url)

更严谨的方案是用 SQLite。

11.4 SQLite 存储版本

如果你希望去重更稳，可以直接用 SQLite：

# green_cert_crawler/sqlite_storage.py

import sqlite3
from pathlib import Path
from typing import Iterable

from .storage import dedupe_key
from .models import GreenProductRecord

class SQLiteStore:
def __init__(self, db_path: Path) –> None:
self.db_path = db_path
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self.conn = sqlite3.connect(str(db_path))
self.init_table()

def init_table(self) –> None:
sql = """
CREATE TABLE IF NOT EXISTS green_products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
dedupe_key TEXT UNIQUE,
product_name TEXT,
brand TEXT,
cert_category TEXT,
valid_until TEXT,
link TEXT,
cert_no TEXT,
source_page INTEGER,
fetched_at TEXT,
row_hash TEXT
)
"""
self.conn.execute(sql)
self.conn.commit()

def upsert(self, record: GreenProductRecord) –> bool:
key = dedupe_key(record)

sql = """
INSERT OR IGNORE INTO green_products (
dedupe_key,
product_name,
brand,
cert_category,
valid_until,
link,
cert_no,
source_page,
fetched_at,
row_hash
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""

cursor = self.conn.execute(
sql,
(
key,
record.product_name,
record.brand,
record.cert_category,
record.valid_until,
record.link,
record.cert_no,
record.source_page,
record.fetched_at,
record.row_hash,
),
)
self.conn.commit()
return cursor.rowcount > 0

def upsert_many(self, records: Iterable[GreenProductRecord]) –> int:
count = 0
for record in records:
if self.upsert(record):
count += 1
return count

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

SQLite 的好处是：

• 断点续跑方便；
• 唯一索引去重可靠；
• 后续查询方便；
• 不需要额外部署服务。

对于个人项目和中小型采集任务，SQLite 是很舒服的选择。

11.5 日志与监控

至少记录：

• 请求 URL；
• 状态码；
• 当前页码；
• 详情链接数量；
• 成功解析数量；
• 失败 URL；
• 去重数量；
• 输出路径。

更进一步可以统计：

success_count
failed_count
duplicate_count
empty_field_count
elapsed_seconds
average_request_time

如果定时运行，可以把统计结果写入一个 summary 文件：

{
"task": "green_product_cert_crawler",
"started_at": "2026-06-09 10:00:00",
"finished_at": "2026-06-09 10:05:30",
"pages": 10,
"parsed": 215,
"saved": 208,
"duplicates": 7,
"failed": 3
}

11.6 定时任务

Linux cron 示例：

0 3 * * * cd /opt/green_cert_crawler && /opt/green_cert_crawler/.venv/bin/python -m green_cert_crawler.main –start-page 1 –end-page 20 >> cron.log 2>&1

含义：每天凌晨 3 点执行一次。

如果任务更复杂，可以用 Airflow、Prefect 或其他调度系统。但别一开始就上重型调度。先把单次任务跑稳，再谈工程化调度。

11.7 变更监控

认证目录抓下来之后，可以和上一次结果比较：

• 新增证书；
• 删除或下架记录；
• 有效期变化；
• 品牌字段变化；
• 认证类别变化。

简单比较逻辑：

import csv

def load_csv(path):
rows = {}
with open(path, "r", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
for row in reader:
key = row.get("认证编号") or row.get("链接")
if key:
rows[key] = row
return rows

old = load_csv("data/output/green_products_old.csv")
new = load_csv("data/output/green_products.csv")

added = set(new) – set(old)
removed = set(old) – set(new)
common = set(old) & set(new)

changed = []
for key in common:
if old[key] != new[key]:
changed.append(key)

print("新增", len(added))
print("移除", len(removed))
print("变更", len(changed))

这种增量对比非常有用。很多时候，数据的变化比数据本身更值得关注。

11.8 Playwright 兜底方案

如果页面必须动态渲染，可以用 Playwright。

安装：

pip install playwright
playwright install chromium

示例：

from playwright.sync_api import sync_playwright

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

但是 Playwright 不是万能钥匙。它适合解决“页面需要 JS 渲染”的问题，不适合拿来绕过站点限制。能抓接口就抓接口，能静态解析就静态解析，最后才考虑浏览器渲染。

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

这篇文章完成了一个绿色产品认证目录分页爬虫的完整设计和实现。

我们做了这些事情：

我写这类爬虫时有一个习惯：先别急着抓全站。先抓 1 页，保存 HTML；再抓 3 条详情，保存样本；然后把解析逻辑写到本地 fixture 能稳定通过。只有当解析层稳定了，再去跑分页。这样看起来慢一点，但最后返工会少很多。

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

第一，把项目改造成 Scrapy。

Scrapy 更适合大规模目录页，内置请求调度、去重、失败重试、Pipeline 和中间件。当前代码的 fetcher、parser、storage 思路都可以迁移过去。

第二，接入数据库和搜索。

如果数据量增加，可以把 SQLite 换成 MySQL 或 PostgreSQL，再把产品名、品牌、认证类别做索引，甚至同步到 Elasticsearch 或 Meilisearch 做检索。

第三，做变更监控和可视化。

认证数据最有价值的地方往往不是“某一天抓到了什么”，而是“相比上一次发生了什么变化”。新增、过期、换证、类别变更，都可以变成监控指标。

技术分享写到这里，代码已经不算短了，但逻辑并不复杂。爬虫真正难的地方，很多时候不是请求网页，而是把边界处理好：合法合规、频率克制、字段容错、去重清楚、结果可复查。把这些做好，一个目录型采集任务才算真的能用。

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

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

✅ 免责声明

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

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

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