Python 实战：抓取产品召回公告目录分页，并进入详情页提取产品名、品牌、召回时间、原因摘要与详情链接

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

全文目录：

• • 🌟 开篇语
  • 0️⃣ 前言（Preface）
  • 1️⃣ 摘要（Abstract）
  • 2️⃣ 背景与需求（Why）
  • • 为什么要爬产品召回公告？
    • 目标站点与目标字段
  • 3️⃣ 合规与注意事项（必写）
  • • robots.txt 基本说明
    • 频率控制，不要攻击式并发
    • 不采集敏感信息，不绕过限制
    • 数据使用保持中性
  • 4️⃣ 技术选型与整体流程（What/How）
  • • 静态、动态还是 API？
    • 整体流程
    • 为什么选 requests + BeautifulSoup + SQLite？
  • 5️⃣ 环境准备与依赖安装（可复现）
  • • Python 版本
    • 安装依赖
    • 推荐项目结构
  • 6️⃣ 核心实现：请求层（Fetcher）
  • • headers：UA 与 Referer
    • timeout：不要无限等待
    • session/cookie：复用连接
    • 失败处理：重试与退避
    • 代码：settings.py
    • 代码：fetcher.py
  • 7️⃣ 核心实现：解析层（Parser）
  • • 解析方式
    • 列表页如何拿详情链接
    • 详情页如何抽字段
    • 缺失字段怎么办？
    • 代码：utils.py
    • 代码：models.py
    • 代码：parser.py
  • 8️⃣ 数据存储与导出（Storage）
  • • 起步选择：SQLite + CSV
    • 字段映射表
    • 去重策略
    • 代码：storage.py
  • 9️⃣ 运行方式与结果展示（必写）
  • • 代码：main.py
    • 如何启动
    • 输出在哪里
    • 示例结果展示
  • 🔟 常见问题与排错（强烈建议写）
  • • 1. 遇到 403 怎么办？
    • 2. 遇到 429 怎么办？
    • 3. HTML 抓到空壳怎么办？
    • 4. 解析报错怎么办？
    • • 字段不存在
      • 选择器变化
      • 列表页混入非详情链接
    • 5. 编码或乱码如何处理？
    • 6. 为什么产品名或品牌有时不准？
    • 7. 抓取结果重复怎么办？
    • 8. 日期格式不统一怎么办？
    • 9. 详情页字段太长怎么办？
  • 1️⃣1️⃣ 进阶优化（可选但加分）
  • • 1. 并发优化
    • 2. 断点续跑
    • 3. 内容变更检测
    • 4. 日志与监控
    • 5. 定时任务
    • 6. 切换到 Scrapy
    • 7. 切换到 Playwright
  • 1️⃣2️⃣ 总结与延伸阅读
  • 🌟 文末
  • • ✅ 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集
    • ✅ 免责声明

🌟 开篇语

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

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

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

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

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

0️⃣ 前言（Preface）

这篇文章要爬取的是一个产品召回公告目录，用 Python 的 requests + BeautifulSoup + SQLite/CSV 完成分页列表抓取、详情页解析、字段清洗和结果导出，最终产出一份可用于检索、分析和长期归档的产品召回数据表。

读完你可以获得：

这不是一篇炫技文章。我更希望它像一次真实开发记录：先跑通最小闭环，再慢慢补请求层、解析层、容错、存储、日志和运行方式。召回公告这类页面看起来不复杂，但做过的人都知道，真正麻烦的地方通常不在请求，而在字段不统一、页面结构变化、列表与详情字段重复、缺失值和去重策略。

1️⃣ 摘要（Abstract）

本文以产品召回公告目录为目标，使用 Python、requests、BeautifulSoup、lxml 和 SQLite 实现一个可复现的分页爬虫：从目录页逐页提取详情链接，再进入详情页解析产品名、品牌、召回时间、原因摘要和详情链接，并将结果导出为 SQLite 与 CSV 文件。

读完后你会得到：

示例站点选用 CPSC 的公开召回页面。CPSC 的目录页展示召回公告，页面底部也存在分页导航，例如当前页、Page 2、Page 3、Next 等分页入口；详情页中能看到产品名称、风险原因、召回日期、召回详情等字段，适合用来演示真实网页的二段式采集。详情页中也能看到 Name of Product、Hazard、Recall Date 等字段，这些字段会被我们映射为目标数据表中的核心字段。

2️⃣ 背景与需求（Why）

为什么要爬产品召回公告？

产品召回公告属于典型的公开信息聚合场景。它不像评论、电商价格、用户画像那样敏感，更多是围绕公开发布的安全提醒、召回批次、处理方式和风险描述做信息整理。实际工作中，这类数据通常有几种用途：

第一是做信息聚合。 如果你关注某类产品，比如儿童用品、小家电、户外用品、电池类产品，就不太可能每天人工翻多个召回公告网站。爬虫可以定时采集公开目录，把新公告整理到自己的数据库里。

第二是做数据分析。 召回公告里常见字段包括产品名称、品牌或企业名、召回日期、风险类型、处理方式、销售渠道、事故报告、生产地等。经过长期积累之后，可以统计不同品类的风险分布，也可以分析某类产品的召回高发原因。

第三是做自动化提醒。 例如一个跨境电商团队希望监控自己经营类目中是否出现相关召回公告；或者一个数据团队想把召回公告接入 BI 报表。爬虫可以作为最前面的采集层，把网页里的非结构化文字变成结构化记录。

我平时写这类爬虫，会先把目标拆成两个问题：

不要试图只在列表页解决一切。很多公告网站列表页只给标题、日期和原因摘要，字段完整性通常在详情页里。这个案例的实战重点，也正是“列表页 + 详情页”的二段式抓取。

目标站点与目标字段

本文以公开产品召回目录为示例目标，字段设计如下：

为什么额外加 source_title、content_hash 和 crawled_at？因为真实项目里只存五个字段有点不够用。详情链接能定位原文，标题能辅助人工核对，哈希能判断内容是否变化，抓取时间能帮助排查数据是不是过期。这些字段不会破坏需求，反而能让项目更像生产环境。

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

爬虫不是“能抓到就可以随便抓”。越是公开页面，越应该保持克制。我的习惯是先做三件事：看 robots.txt、看是否有公开 API、控制访问频率。

robots.txt 基本说明

robots.txt 是网站在根路径下提供给爬虫读取的访问约定文件。它不是登录授权，也不是万能法律许可，但它表达了站点对自动化访问的基本边界。开发爬虫前，建议先访问：

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

在 Python 中可以用标准库 urllib.robotparser 做基础判断。本文代码会在启动时提供一个 check_robots() 示例。需要注意的是，不同站点对不同 User-Agent 可能有不同规则，所以不要只看一眼就下结论，最好在代码中显式配置自己的 User-Agent。

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

召回公告页面通常不是高频变化的数据。没有必要每秒打几十个请求，也没有必要上来就多线程冲满。本文默认策略是：

这类数据做长期采集，稳定比快重要。我宁愿一分钟只抓几十条，也不愿意因为鲁莽请求被封掉，或者给别人服务器制造负担。

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

本文只采集公开召回公告页面中的公开字段，不采集个人信息，不绕过登录限制，不绕过付费限制，不破解接口签名，不模拟用户隐私操作。

如果站点提供公开 API，优先考虑 API；如果教学目标是训练网页解析，也应该控制范围，只抓少量公开页面。CPSC 官方页面说明其召回信息也提供机器可读的 API 访问方式，并支持 XML 与 JSON，这意味着在真实业务里可以优先评估 API 是否更稳定。但本文为了训练“列表页 + 详情页”二段式解析，仍然以网页抓取作为主线。

数据使用保持中性

召回公告属于公共安全信息。本文目的只是技术学习和公开信息整理，不用于误导性宣传，不做产品好坏排名，不对企业作额外推断。数据结论应该回到原始公告，不要用爬虫结果替代权威说明。

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

静态、动态还是 API？

常见网页数据源大概有三类：

第一类是静态 HTML。 服务端直接返回包含数据的 HTML，requests 请求后就能在响应文本里看到目标字段。这类最适合用 requests + BeautifulSoup/lxml。

第二类是动态渲染页面。 初始 HTML 只是空壳，数据由浏览器执行 JavaScript 后再渲染。遇到这种情况，要么找接口，要么用 Playwright/Selenium 渲染页面。

第三类是公开 API。 数据直接以 JSON 或 XML 返回。真实业务里 API 往往更稳定，但学习网页解析时，HTML 解析仍然很有价值。

本文案例属于“以静态页面抓取为主，同时知道存在 API”的类型。目录页中能看到公告文本和分页入口，详情页中也能看到字段文本，因此可以用 requests 抓 HTML，再用 BeautifulSoup 解析。CPSC 详情页示例中直接展示了 Name of Product、Hazard、Recall Date、Recall Details 等信息，适合静态解析。

整体流程

本文爬虫流程如下：

启动程序
↓
读取配置
↓
检查 robots.txt
↓
请求召回目录第一页
↓
解析详情页链接
↓
进入每个详情页
↓
解析产品名、品牌、召回时间、原因摘要、详情链接
↓
清洗字段、补默认值、生成 hash
↓
写入 SQLite
↓
导出 CSV
↓
打印抓取统计

更简洁地说，就是：

采集 → 解析 → 清洗 → 存储

为什么选 requests + BeautifulSoup + SQLite？

我没有一上来就用 Scrapy，原因很简单：这篇文章的重点不是框架，而是“目录分页 + 详情页字段抽取”的基本功。对于中小规模公告采集，requests + BeautifulSoup 足够清楚，代码量也更适合教学。

选择理由：

什么时候换 Scrapy？当页面数量变多、需要调度队列、失败重试、去重、增量抓取、日志统计和中间件时，可以换。

什么时候换 Playwright？当 HTML 里拿不到数据、页面依赖浏览器渲染、接口签名复杂或必须等待异步数据加载时，可以换。

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

Python 版本

推荐使用：

Python 3.10+

我建议至少使用 Python 3.10，因为类型标注、路径处理和依赖兼容性都比较舒服。如果你用 Python 3.11 或 3.12，也没有问题。

安装依赖

创建虚拟环境：

python -m venv .venv

Windows 激活：

.venv\\Scripts\\activate

macOS / Linux 激活：

source .venv/bin/activate

安装依赖：

pip install requests beautifulsoup4 lxml python-dateutil tqdm

生成 requirements.txt：

requests>=2.31.0
beautifulsoup4>=4.12.0
lxml>=5.0.0
python-dateutil>=2.9.0
tqdm>=4.66.0

安装：

pip install -r requirements.txt

推荐项目结构

recall_spider/
├── README.md
├── requirements.txt
├── main.py
├── recall_spider/
│ ├── __init__.py
│ ├── settings.py
│ ├── models.py
│ ├── fetcher.py
│ ├── parser.py
│ ├── storage.py
│ └── utils.py
└── output/
├── recalls.db
└── recalls.csv

这个结构不复杂，但已经能把职责拆开：

• settings.py：配置。
• models.py：数据模型。
• fetcher.py：请求层。
• parser.py：解析层。
• storage.py：存储层。
• utils.py：清洗、哈希、日期处理等工具。
• main.py：程序入口。

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

请求层的任务不是“把页面拿回来”这么简单。一个靠谱的 Fetcher 至少要考虑：

headers：UA 与 Referer

很多网站会对默认 Python UA 不太友好。我们应该设置一个清楚、克制的 User-Agent，告诉对方这不是浏览器攻击流量，而是一个学习或研究用途的低频访问程序。

示例：

HEADERS = {
"User-Agent": (
"RecallResearchBot/1.0 "
"(educational project; contact: example@example.com)"
),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Referer": "https://www.cpsc.gov/",
}

真实项目里，contact 最好写真实联系邮箱。不要伪装成搜索引擎蜘蛛，也不要滥用浏览器 UA 去绕过限制。

timeout：不要无限等待

requests.get() 如果不设置 timeout，网络异常时可能卡住很久。推荐使用连接超时和读取超时的元组：

TIMEOUT = (5, 20)

含义是：连接最多等 5 秒，读取最多等 20 秒。

session/cookie：复用连接

公告类网站一般不需要登录 cookie，但 requests.Session() 仍然有价值。它能复用连接，减少握手开销，也方便统一 headers。

失败处理：重试与退避

失败重试不能简单写成“失败就立刻重试三次”。如果对方返回 429，说明你请求太快；如果 503，说明服务暂时不可用。这时应该等待，而且等待时间最好逐步变长。

策略：

第 1 次失败：等待 2 秒左右
第 2 次失败：等待 4 秒左右
第 3 次失败：等待 8 秒左右
再失败：记录错误，跳过或退出

再加一点随机抖动，可以避免多个请求同时恢复。

代码：settings.py

# recall_spider/settings.py
from pathlib import Path

BASE_URL = "https://www.cpsc.gov"
START_URL = "https://www.cpsc.gov/Recalls"

PROJECT_ROOT = Path(__file__).resolve().parent.parent
OUTPUT_DIR = PROJECT_ROOT / "output"
SQLITE_PATH = OUTPUT_DIR / "recalls.db"
CSV_PATH = OUTPUT_DIR / "recalls.csv"

HEADERS = {
"User-Agent": (
"RecallResearchBot/1.0 "
"(educational project; contact: example@example.com)"
),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Referer": BASE_URL,
}

TIMEOUT = (5, 20)

MAX_RETRIES = 3
BACKOFF_FACTOR = 2.0

LIST_DELAY_RANGE = (1.2, 2.8)
DETAIL_DELAY_RANGE = (0.8, 2.2)

DEFAULT_MAX_PAGES = 3

代码：fetcher.py

# recall_spider/fetcher.py
from __future__ import annotations

import random
import time
from typing import Optional

import requests

from recall_spider.settings import (
HEADERS,
TIMEOUT,
MAX_RETRIES,
BACKOFF_FACTOR,
)

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

class Fetcher:
def __init__(self) –> None:
self.session = requests.Session()
self.session.headers.update(HEADERS)

def get(self, url: str, referer: Optional[str] = None) –> str:
headers = dict(HEADERS)
if referer:
headers["Referer"] = referer

last_error: Optional[Exception] = None

for attempt in range(1, MAX_RETRIES + 1):
try:
response = self.session.get(
url,
headers=headers,
timeout=TIMEOUT,
)

if response.status_code in {429, 500, 502, 503, 504}:
wait_seconds = self._calc_backoff(attempt)
print(
f"[WARN] status={response.status_code}, "
f"attempt={attempt}/{MAX_RETRIES}, "
f"sleep={wait_seconds:.1f}s, url={url}"
)
time.sleep(wait_seconds)
continue

response.raise_for_status()

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

return response.text

except requests.RequestException as exc:
last_error = exc
wait_seconds = self._calc_backoff(attempt)
print(
f"[WARN] request failed, attempt={attempt}/{MAX_RETRIES}, "
f"sleep={wait_seconds:.1f}s, url={url}, error={exc}"
)
time.sleep(wait_seconds)

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

@staticmethod
def _calc_backoff(attempt: int) –> float:
base = BACKOFF_FACTOR ** attempt
jitter = random.uniform(0.2, 1.2)
return base + jitter

这段 Fetcher 不追求花哨，但足够实用。状态码 429、5xx 会退避重试；普通请求异常也会重试。真正失败后抛出 FetchError，由主程序决定跳过还是终止。

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

解析层是这类项目最容易翻车的地方。原因有三个：

所以我会把解析分成两部分：

parse_list_page() # 从列表页提取详情链接
parse_detail_page() # 从详情页提取目标字段

解析方式

本文使用 BeautifulSoup，并指定 lxml 解析器：

soup = BeautifulSoup(html, "lxml")

选择 CSS 选择器和文本规则混合解析。为什么不用纯 XPath？因为公告页面中的字段标签有时是普通文本节点，不一定全部有稳定 class。BeautifulSoup 做“按文本行找字段标签”会更顺手。

列表页如何拿详情链接

目录页的核心目标只有一个：找到详情页 URL。

在示例站点中，召回详情页 URL 通常包含：

/Recalls/年份/

例如：

/Recalls/2026/World-Bright-International-Limited-Recalls-…

因此我们可以遍历所有 <a href="">，用正则筛选 /Recalls/\\d{4}/。目录页底部的分页链接可以通过 “Next” 文本或分页 class 找到。CPSC 目录页底部能看到 Page 2、Page 3、Next 等分页项，说明它适合演示目录分页抓取。

详情页如何抽字段

详情页中常见结构大致是：

Name of Product:
Agio Menlo Woven Patio Swings

Hazard:
The swing seat can detach…

Recall Date:
May 14, 2026

示例详情页中可以直接看到 Name of Product、Hazard、Remedy、Recall Date、Units 等字段。

解析策略：

缺失字段怎么办？

召回公告里并不一定有明确的 “Brand” 标签。这个时候不要硬编。本文将品牌定义为“品牌/企业主体”，抽取优先级如下：

这叫“可解释兜底”。爬虫最忌讳的是把猜测包装成确定值。字段可以推断，但要让后续使用者知道它是怎么来的。

代码：utils.py

# recall_spider/utils.py
from __future__ import annotations

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

from dateutil import parser as date_parser

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

def normalize_lines(lines: Iterable[str]) –> list[str]:
result: list[str] = []
for line in lines:
clean = normalize_space(line)
if clean:
result.append(clean)
return result

def parse_date_to_iso(text: str) –> str:
text = normalize_space(text)
if not text:
return ""

try:
dt = date_parser.parse(text)
return dt.date().isoformat()
except (ValueError, TypeError, OverflowError):
return text

def short_text(text: str, max_len: int = 240) –> str:
text = normalize_space(text)
if len(text) <= max_len:
return text
return text[: max_len – 3].rstrip() + "…"

def sha256_text(text: str) –> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()

def now_utc_iso() –> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()

def first_non_empty(*values: str) –> str:
for value in values:
clean = normalize_space(value)
if clean:
return clean
return ""

代码：models.py

# recall_spider/models.py
from __future__ import annotations

from dataclasses import dataclass, asdict

@dataclass(slots=True)
class RecallItem:
product_name: str
brand: str
recall_date: str
reason_summary: str
detail_url: str
source_title: str
content_hash: str
crawled_at: str

def to_dict(self) –> dict[str, str]:
return asdict(self)

代码：parser.py

# recall_spider/parser.py
from __future__ import annotations

import re
from urllib.parse import urljoin

from bs4 import BeautifulSoup

from recall_spider.models import RecallItem
from recall_spider.settings import BASE_URL
from recall_spider.utils import (
first_non_empty,
normalize_lines,
normalize_space,
now_utc_iso,
parse_date_to_iso,
sha256_text,
short_text,
)

DETAIL_URL_PATTERN = re.compile(r"/Recalls/\\d{4}/", re.IGNORECASE)

FIELD_LABELS = {
"name_of_product": ["Name of Product", "Product Name"],
"hazard": ["Hazard", "Risk", "Reason"],
"recall_date": ["Recall Date", "Date"],
"description": ["Description"],
"brand": ["Brand"],
"recall_number": ["Recall number", "Recall Number"],
"importers": ["Importer(s)", "Importer"],
"distributors": ["Distributor(s)", "Distributor"],
"manufacturers": ["Manufacturer(s)", "Manufacturer"],
}

STOP_LABELS = {
"Name of Product",
"Product Name",
"Hazard",
"Risk",
"Reason",
"Remedy",
"Recall Date",
"Date",
"Units",
"Consumer Contact",
"Recall Details",
"Description",
"Incidents/Injuries",
"Sold At",
"Importer(s)",
"Distributor(s)",
"Manufacturer(s)",
"Manufactured In",
"Recall number",
"Recall Number",
"Related Recalls",
"About the U.S. CPSC",
}

def parse_list_page(html: str, current_url: str) –> tuple[list[str], str]:
soup = BeautifulSoup(html, "lxml")

detail_urls: list[str] = []
seen: set[str] = set()

for a_tag in soup.select("a[href]"):
href = a_tag.get("href", "")
text = normalize_space(a_tag.get_text(" ", strip=True))

if not href:
continue

if not DETAIL_URL_PATTERN.search(href):
continue

if not text:
continue

absolute_url = urljoin(BASE_URL, href)

if absolute_url in seen:
continue

seen.add(absolute_url)
detail_urls.append(absolute_url)

next_url = parse_next_page_url(soup, current_url)

return detail_urls, next_url

def parse_next_page_url(soup: BeautifulSoup, current_url: str) –> str:
candidates = [
"li.pager__item–next a[href]",
"a[rel='next'][href]",
"a[title='Go to next page'][href]",
]

for selector in candidates:
node = soup.select_one(selector)
if node and node.get("href"):
return urljoin(current_url, node["href"])

for a_tag in soup.select("a[href]"):
text = normalize_space(a_tag.get_text(" ", strip=True)).lower()
if text in {"next", "next ›", "›", "下一页"} or "next" in text:
return urljoin(current_url, a_tag["href"])

return ""

def parse_detail_page(html: str, detail_url: str) –> RecallItem:
soup = BeautifulSoup(html, "lxml")

title = extract_title(soup)
lines = extract_main_lines(soup)

product_name = first_non_empty(
extract_field(lines, FIELD_LABELS["name_of_product"]),
guess_product_from_title(title),
)

hazard = extract_field(lines, FIELD_LABELS["hazard"])
description = extract_field(lines, FIELD_LABELS["description"])
reason_summary = short_text(first_non_empty(hazard, description, title), 260)

recall_date_raw = extract_field(lines, FIELD_LABELS["recall_date"])
recall_date = parse_date_to_iso(recall_date_raw)

brand = first_non_empty(
extract_field(lines, FIELD_LABELS["brand"]),
guess_company_from_title(title),
guess_brand_from_product(product_name),
)

content_basis = "|".join(
[
product_name,
brand,
recall_date,
reason_summary,
detail_url,
title,
]
)

return RecallItem(
product_name=product_name,
brand=brand,
recall_date=recall_date,
reason_summary=reason_summary,
detail_url=detail_url,
source_title=title,
content_hash=sha256_text(content_basis),
crawled_at=now_utc_iso(),
)

def extract_title(soup: BeautifulSoup) –> str:
h1 = soup.find("h1")
if h1:
return normalize_space(h1.get_text(" ", strip=True))

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

return ""

def extract_main_lines(soup: BeautifulSoup) –> list[str]:
for tag_name in ["script", "style", "noscript", "svg"]:
for node in soup.find_all(tag_name):
node.decompose()

main = soup.find("main") or soup.find("article") or soup.body or soup

text = main.get_text("\\n", strip=True)
lines = normalize_lines(text.splitlines())

return lines

def extract_field(lines: list[str], label_candidates: list[str]) –> str:
normalized_labels = {label.lower().rstrip(":") for label in label_candidates}

for index, line in enumerate(lines):
clean_line = normalize_space(line)
lower_line = clean_line.lower().rstrip(":")

if lower_line in normalized_labels:
return collect_value_after_label(lines, index)

for label in label_candidates:
prefix = label + ":"
if clean_line.lower().startswith(prefix.lower()):
value = clean_line[len(prefix):].strip()
if value:
return value
return collect_value_after_label(lines, index)

return ""

def collect_value_after_label(lines: list[str], label_index: int) –> str:
values: list[str] = []

for next_line in lines[label_index + 1:]:
clean = normalize_space(next_line)
if not clean:
continue

if clean.rstrip(":") in STOP_LABELS:
break

if is_probably_section_title(clean):
break

values.append(clean)

if len(" ".join(values)) >= 500:
break

return normalize_space(" ".join(values))

def is_probably_section_title(text: str) –> bool:
if text in STOP_LABELS:
return True

if text.endswith(":") and len(text) <= 40:
return True

return False

def guess_company_from_title(title: str) –> str:
title = normalize_space(title)
if not title:
return ""

patterns = [
r"^(.*?)\\s+Recalls?\\s+",
r"^(.*?)\\s+Recall[s]?:\\s+",
r"^(.*?)\\s+Announces?\\s+Recall",
]

for pattern in patterns:
match = re.search(pattern, title, flags=re.IGNORECASE)
if match:
candidate = normalize_space(match.group(1))
candidate = cleanup_brand_candidate(candidate)
if candidate:
return candidate

if title.lower().startswith("cpsc warns consumers to stop using "):
rest = title[len("CPSC Warns Consumers to Stop Using "):]
return guess_brand_from_product(rest)

return ""

def guess_product_from_title(title: str) –> str:
title = normalize_space(title)
if not title:
return ""

match = re.search(
r"Recalls?\\s+(.*?)\\s+Due to\\s+",
title,
flags=re.IGNORECASE,
)
if match:
return normalize_space(match.group(1))

match = re.search(
r"Stop Using\\s+(.*?)\\s+Immediately",
title,
flags=re.IGNORECASE,
)
if match:
return normalize_space(match.group(1))

return ""

def guess_brand_from_product(product_name: str) –> str:
product_name = normalize_space(product_name)
if not product_name:
return ""

words = product_name.split()
if not words:
return ""

first = words[0].strip(",;:()[]{}")
if len(first) <= 1:
return ""

return first

def cleanup_brand_candidate(text: str) –> str:
text = normalize_space(text)

noise_prefixes = [
"CPSC Warns Consumers to Stop Using",
"The U.S. Consumer Product Safety Commission",
]

for prefix in noise_prefixes:
if text.lower().startswith(prefix.lower()):
text = text[len(prefix):].strip()

return text.strip(" -–—:;,.")

这份解析器的重点不是选择器多精准，而是“容错思路”清楚。列表页用链接规则筛详情 URL，详情页用字段标签提取内容。页面 class 变了，文本标签通常还在；如果文本标签也变了，就把 FIELD_LABELS 补充一下。

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

起步选择：SQLite + CSV

我建议这个项目起步用 SQLite。原因很现实：

字段映射表

示例中的 Agio Menlo Woven Patio Swings 详情页确实包含产品名称、风险原因、召回日期、召回单位数量、联系方式和召回详情等字段；我们的字段表只抽取其中最核心的一部分，便于通用化。

去重策略

本文使用两层去重：

第一层：详情链接唯一。 同一个公告详情 URL 只保留一条。

第二层：内容 hash。 当详情页内容变化时，hash 会变化。本文示例中使用 ON CONFLICT(detail_url) DO UPDATE，也就是说同一个 URL 再次抓到时会更新内容，适合做轻量增量。

如果你需要保留历史版本，可以设计一个 recall_history 表，把每次变更都插入新记录。

代码：storage.py

# recall_spider/storage.py
from __future__ import annotations

import csv
import sqlite3
from pathlib import Path
from typing import Iterable

from recall_spider.models import RecallItem
from recall_spider.settings import CSV_PATH, SQLITE_PATH

CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS recalls (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_name TEXT NOT NULL DEFAULT '',
brand TEXT NOT NULL DEFAULT '',
recall_date TEXT NOT NULL DEFAULT '',
reason_summary TEXT NOT NULL DEFAULT '',
detail_url TEXT NOT NULL UNIQUE,
source_title TEXT NOT NULL DEFAULT '',
content_hash TEXT NOT NULL DEFAULT '',
crawled_at TEXT NOT NULL DEFAULT ''
);
"""

CREATE_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_recalls_recall_date
ON recalls(recall_date);
"""

UPSERT_SQL = """
INSERT INTO recalls (
product_name,
brand,
recall_date,
reason_summary,
detail_url,
source_title,
content_hash,
crawled_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(detail_url) DO UPDATE SET
product_name=excluded.product_name,
brand=excluded.brand,
recall_date=excluded.recall_date,
reason_summary=excluded.reason_summary,
source_title=excluded.source_title,
content_hash=excluded.content_hash,
crawled_at=excluded.crawled_at;
"""

class RecallStorage:
def __init__(self, db_path: Path = SQLITE_PATH) –> None:
self.db_path = db_path
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self.connection = sqlite3.connect(self.db_path)
self.connection.execute(CREATE_TABLE_SQL)
self.connection.execute(CREATE_INDEX_SQL)
self.connection.commit()

def save_item(self, item: RecallItem) –> None:
self.connection.execute(
UPSERT_SQL,
(
item.product_name,
item.brand,
item.recall_date,
item.reason_summary,
item.detail_url,
item.source_title,
item.content_hash,
item.crawled_at,
),
)
self.connection.commit()

def save_many(self, items: Iterable[RecallItem]) –> None:
with self.connection:
for item in items:
self.connection.execute(
UPSERT_SQL,
(
item.product_name,
item.brand,
item.recall_date,
item.reason_summary,
item.detail_url,
item.source_title,
item.content_hash,
item.crawled_at,
),
)

def export_csv(self, csv_path: Path = CSV_PATH) –> None:
csv_path.parent.mkdir(parents=True, exist_ok=True)

cursor = self.connection.execute(
"""
SELECT
product_name,
brand,
recall_date,
reason_summary,
detail_url,
source_title,
content_hash,
crawled_at
FROM recalls
ORDER BY recall_date DESC, id DESC;
"""
)

fieldnames = [
"product_name",
"brand",
"recall_date",
"reason_summary",
"detail_url",
"source_title",
"content_hash",
"crawled_at",
]

with csv_path.open("w", encoding="utf-8-sig", newline="") as file:
writer = csv.writer(file)
writer.writerow(fieldnames)
writer.writerows(cursor.fetchall())

def count(self) –> int:
cursor = self.connection.execute("SELECT COUNT(*) FROM recalls;")
row = cursor.fetchone()
return int(row[0] if row else 0)

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

CSV 使用 utf-8-sig，是为了兼容 Excel。很多人第一次导出 CSV 后发现中文乱码，其实不是数据错，而是 Excel 对 UTF-8 的识别不稳定。加 BOM 是一个实用折中。

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

代码：main.py

# main.py
from __future__ import annotations

import argparse
import random
import time
import urllib.robotparser
from urllib.parse import urlparse

from tqdm import tqdm

from recall_spider.fetcher import FetchError, Fetcher
from recall_spider.parser import parse_detail_page, parse_list_page
from recall_spider.settings import (
BASE_URL,
CSV_PATH,
DETAIL_DELAY_RANGE,
LIST_DELAY_RANGE,
SQLITE_PATH,
START_URL,
DEFAULT_MAX_PAGES,
)
from recall_spider.storage import RecallStorage

def check_robots(start_url: str, user_agent: str = "*") –> bool:
parsed = urlparse(start_url)
robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"

rp = urllib.robotparser.RobotFileParser()
rp.set_url(robots_url)

try:
rp.read()
except Exception as exc:
print(f"[WARN] robots.txt check failed: {exc}")
return True

allowed = rp.can_fetch(user_agent, start_url)
if not allowed:
print(f"[ERROR] robots.txt does not allow fetching: {start_url}")
else:
print(f"[INFO] robots.txt allows fetching start url: {start_url}")

return allowed

def sleep_random(delay_range: tuple[float, float], label: str) –> None:
seconds = random.uniform(*delay_range)
print(f"[INFO] sleep {seconds:.2f}s after {label}")
time.sleep(seconds)

def run(max_pages: int, start_url: str, obey_robots: bool) –> None:
if obey_robots and not check_robots(start_url):
return

fetcher = Fetcher()
storage = RecallStorage(SQLITE_PATH)

current_url = start_url
visited_list_pages: set[str] = set()
visited_detail_urls: set[str] = set()

total_links = 0
saved_count = 0
failed_count = 0

try:
for page_no in range(1, max_pages + 1):
if not current_url:
print("[INFO] no next page url, stop.")
break

if current_url in visited_list_pages:
print(f"[INFO] list page already visited: {current_url}")
break

visited_list_pages.add(current_url)

print(f"\\n[INFO] fetching list page {page_no}: {current_url}")

try:
list_html = fetcher.get(current_url, referer=BASE_URL)
except FetchError as exc:
print(f"[ERROR] failed to fetch list page: {exc}")
break

detail_urls, next_url = parse_list_page(list_html, current_url)
total_links += len(detail_urls)

print(
f"[INFO] page={page_no}, detail_links={len(detail_urls)}, "
f"next_url={next_url or '-'}"
)

for detail_url in tqdm(detail_urls, desc=f"details page {page_no}"):
if detail_url in visited_detail_urls:
continue

visited_detail_urls.add(detail_url)

try:
detail_html = fetcher.get(detail_url, referer=current_url)
item = parse_detail_page(detail_html, detail_url)
storage.save_item(item)
saved_count += 1

except Exception as exc:
failed_count += 1
print(f"[ERROR] detail failed: url={detail_url}, error={exc}")

sleep_random(DETAIL_DELAY_RANGE, "detail page")

current_url = next_url
sleep_random(LIST_DELAY_RANGE, "list page")

storage.export_csv(CSV_PATH)

print("\\n[DONE] crawl finished.")
print(f"[DONE] total links found: {total_links}")
print(f"[DONE] saved items: {saved_count}")
print(f"[DONE] failed details: {failed_count}")
print(f"[DONE] sqlite: {SQLITE_PATH}")
print(f"[DONE] csv: {CSV_PATH}")
print(f"[DONE] rows in database: {storage.count()}")

finally:
storage.close()

def parse_args() –> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Crawl product recall directory pages and detail pages."
)

parser.add_argument(
"–max-pages",
type=int,
default=DEFAULT_MAX_PAGES,
help="Maximum list pages to crawl.",
)

parser.add_argument(
"–start-url",
type=str,
default=START_URL,
help="Start URL of recall directory.",
)

parser.add_argument(
"–ignore-robots",
action="store_true",
help="Skip robots.txt check. Not recommended.",
)

return parser.parse_args()

if __name__ == "__main__":
args = parse_args()
run(
max_pages=args.max_pages,
start_url=args.start_url,
obey_robots=not args.ignore_robots,
)

如何启动

在项目根目录运行：

python main.py –max-pages 2

如果你只想试一页：

python main.py –max-pages 1

输出在哪里

默认输出：

output/recalls.db
output/recalls.csv

你可以用 SQLite 命令查看：

sqlite3 output/recalls.db

进入后执行：

SELECT product_name, brand, recall_date, reason_summary, detail_url
FROM recalls
ORDER BY recall_date DESC
LIMIT 5;

也可以直接用 Pandas：

import pandas as pd

df = pd.read_csv("output/recalls.csv")
print(df.head())

示例结果展示

下面是按本文字段设计整理后的示例结果格式。示例值来自公开召回详情页和目录页中可见的信息：例如 Agio Menlo Woven Patio Swings 的详情页展示了产品名、危害、召回日期等字段；目录页也展示了多条公告标题、风险摘要和日期。

注意，品牌字段是启发式提取字段。对于某些公告，标题里的主体可能是进口商、分销商或发布企业，不一定等同于消费者理解里的商标品牌。所以在正式业务中，我会把字段名写成 brand_or_company，或者额外增加 brand_confidence、company_name、product_brand 三个字段，把推断层次拆开。

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

1. 遇到 403 怎么办？

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

处理建议：

第一，先检查合规。 如果 robots 或站点条款不允许采集，不要继续。

第二，降低频率。 把延迟从 1 秒调到 5 秒甚至 10 秒。

第三，补充正常 headers。 不是为了伪装攻击，而是让请求更接近普通 HTTP 客户端的完整表达。

第四，优先找公开 API。 如果站点明确提供 API，直接用 API 通常更稳定。

不要把 403 简单理解成“加代理就好了”。代理只能改变网络出口，不能改变访问边界。技术上能绕和应该绕，是两回事。

2. 遇到 429 怎么办？

429 的意思通常是请求过多。正确做法是降速，而不是加并发。

建议：

if response.status_code == 429:
time.sleep(60)

更稳妥的是读取响应头中的 Retry-After：

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

本文 Fetcher 已经对 429 做了退避重试，但真实项目里可以更精细。

3. HTML 抓到空壳怎么办？

如果 requests.get(url).text 里看不到目标数据，而浏览器里能看到，通常是动态渲染。处理路线有三条：

第一，打开浏览器开发者工具，看 Network。 很多动态页面的数据来自 JSON 接口，找到接口后直接请求接口。

第二，查看页面源码。 如果源码没有数据，说明不是静态 HTML。

第三，使用 Playwright。 让浏览器真实执行 JavaScript，然后再取渲染后的 HTML。

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。浏览器渲染成本高，速度慢，也更容易给站点造成压力。能用 API 或静态 HTML，就不要过度复杂化。

4. 解析报错怎么办？

解析报错一般有几类：

字段不存在

例如：

node.get_text(strip=True)

如果 node 是 None，就会报错。

写法应该改成：

text = node.get_text(strip=True) if node else ""

选择器变化

页面 class 可能改名。解决办法：

本文的 FIELD_LABELS 就是为这个准备的。

列表页混入非详情链接

不是所有 /Recalls 链接都是详情页。可以用正则进一步限制：

DETAIL_URL_PATTERN = re.compile(r"/Recalls/\\d{4}/", re.I)

如果目标站点 URL 规则不同，改正则即可。

5. 编码或乱码如何处理？

优先检查响应编码：

print(response.encoding)
print(response.apparent_encoding)

如果 response.encoding 不准，可以：

response.encoding = response.apparent_encoding

写 CSV 时建议：

encoding="utf-8-sig"

读取时：

pd.read_csv("output/recalls.csv", encoding="utf-8-sig")

6. 为什么产品名或品牌有时不准？

因为公告页面不一定专门提供 “brand” 字段。标题里的主体有时是制造商，有时是进口商，有时是品牌方，有时是监管机构提醒消费者停用某产品。爬虫不是知识图谱，不应该过度推断。

更稳的做法是把字段拆细：

这样后续人工核验会舒服很多。

7. 抓取结果重复怎么办？

重复来源通常有两个：

解决办法：

seen = set()
if url not in seen:
seen.add(url)

数据库层再加唯一约束：

detail_url TEXT NOT NULL UNIQUE

双层去重最稳。内存去重能减少重复请求，数据库唯一索引能兜底。

8. 日期格式不统一怎么办？

英文站点日期可能是：

May 14, 2026

中文站点可能是：

2026年5月14日

如果只处理英文，用 dateutil.parser.parse() 很方便。多语言日期可以自己写规则：

import re

def parse_cn_date(text: str) –> str:
m = re.search(r"(\\d{4})年(\\d{1,2})月(\\d{1,2})日", text)
if not m:
return ""
y, mth, d = m.groups()
return f"{int(y):04d}–{int(mth):02d}–{int(d):02d}"

9. 详情页字段太长怎么办？

原因摘要不宜太长。数据库可以存完整字段，但展示字段应该截断。

本文使用：

short_text(text, max_len=260)

如果你后面要做 NLP 分析，可以额外保存 raw_detail_text，不要只存摘要。

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

1. 并发优化

本文默认串行抓取，适合教学和低频采集。如果数据量变大，可以考虑线程池：

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_one_detail(fetcher, url):
html = fetcher.get(url)
return parse_detail_page(html, url)

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

for future in as_completed(futures):
try:
item = future.result()
storage.save_item(item)
except Exception as exc:
print(exc)

注意，max_workers 不要太大。公告页面不是抢票系统，也不是压测目标。通常 2 到 5 个线程已经足够。

如果想更优雅，可以用 asyncio + httpx，但异步不是银弹。对于低频公告采集，同步代码更容易维护。

2. 断点续跑

目前程序会把详情 URL 存进 SQLite，并通过唯一索引避免重复。但它仍然会请求已经抓过的详情页。可以增加一个判断：

def exists_url(self, detail_url: str) –> bool:
cursor = self.connection.execute(
"SELECT 1 FROM recalls WHERE detail_url = ? LIMIT 1;",
(detail_url,),
)
return cursor.fetchone() is not None

然后在主程序里：

if storage.exists_url(detail_url):
print(f"[INFO] skip existing url: {detail_url}")
continue

不过这里有一个取舍： 如果你永远跳过旧 URL，就无法发现详情页内容更新。产品召回页面有时会更新联系方式或补充处理方式。因此我更喜欢增加一个参数：

python main.py –max-pages 5 –skip-existing

默认更新，必要时跳过。

3. 内容变更检测

当前 content_hash 是根据核心字段生成的。如果你想检测完整详情页变化，可以保存原始正文 hash：

raw_text = "\\n".join(lines)
raw_hash = sha256_text(raw_text)

表结构增加：

raw_hash TEXT NOT NULL DEFAULT ''

如果同一个 URL 的 raw_hash 变化，就说明原文可能更新了。你可以把变化记录到日志或历史表。

4. 日志与监控

教学代码用 print() 没问题，但生产项目建议换 logging：

import logging

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

建议记录：

如果有一天页面结构改了，空字段比例会突然升高，这是很好的报警信号。

5. 定时任务

Linux cron 示例：

0 8 * * * cd /path/to/recall_spider && /path/to/.venv/bin/python main.py –max-pages 3 >> output/cron.log 2>&1

每天上午 8 点抓前三页。召回公告不是秒级数据，一天一次通常够用。

如果你需要更完整的调度、重试、依赖管理，可以考虑 Airflow 或 Prefect。

6. 切换到 Scrapy

当需求变成“抓很多页、很多站点、长期运行”时，Scrapy 更合适。Scrapy 内置：

Scrapy 版本的核心思路不变：

parse() 解析列表页
yield Request(detail_url, callback=parse_detail)

parse_detail() 解析详情页
yield item

结构更工程化，但学习曲线也更高。

7. 切换到 Playwright

如果目标站点改成动态渲染，requests 抓不到数据，可以使用 Playwright。典型结构：

from playwright.sync_api import sync_playwright

def fetch_by_playwright(url: str) –> str:
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
context = browser.new_context(
user_agent=(
"RecallResearchBot/1.0 "
"(educational project; contact: example@example.com)"
)
)
page = context.new_page()
page.goto(url, wait_until="networkidle", timeout=30000)
html = page.content()
browser.close()
return html

但要记住：Playwright 是最后手段之一，不是默认选择。它会消耗更多资源，也更容易被站点视为自动化浏览器访问。

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

这篇文章完成了一个完整的产品召回公告分页爬虫，从目录页开始，逐页发现详情链接，再进入详情页提取产品名、品牌/企业主体、召回日期、原因摘要和详情链接，最后写入 SQLite 并导出 CSV。

回顾一下核心点：

下一步可以继续扩展：

• 改成 Scrapy 项目，加入 AutoThrottle 和 Pipeline。
• 改成 Playwright 版本，适配动态渲染站点。
• 增加 raw_text 字段，做全文检索。
• 增加内容变更历史表，追踪公告更新。
• 接入定时任务，每天自动抓取最新公告。
• 用 Pandas 做召回原因统计、品牌频次统计、月份趋势分析。
• 把 CSV 接入 BI 工具，做内部监控看板。

我个人写这类爬虫时，最看重的不是一次抓了多少条，而是代码过一周、一个月后还能不能稳定跑。公告类页面最怕“今天能跑，明天字段全空”。所以与其追求一开始就大并发，不如先把请求、解析、清洗、存储、日志和排错写扎实。慢一点没关系，稳才是爬虫工程里真正省时间的方式。

🌟 文末

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

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

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

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

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

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

✅ 互动征集

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

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

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

✅ 免责声明

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

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

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