Python爬虫实战：爬取Chrome Web Store扩展商店的分类页面，提取插件名称、评分、下载量、详情链接等关键信息（附 CSV 导出 + JSON格式）！

㊙️本期内容已收录至专栏《Python爬虫实战》，持续完善知识体系与项目实战，建议先订阅收藏，后续查阅更方便～持续更新中！ ㊗️爬虫难度指数：⭐ 🚫声明：本数据&代码仅供学习交流，严禁用于商业用途、倒卖数据或违反目标站点的服务条款等，一切后果皆由使用者本人承担。公开榜单数据一般允许访问，但请务必遵守“君子协议”，技术无罪，责任在人。

全文目录：

• • 🌟 开篇语
  • 1️⃣ 摘要（Abstract）
  • 2️⃣ 背景与需求（Why）
  • • 为什么要爬取插件市场数据？
    • 目标站点与字段清单
  • 3️⃣ 合规与注意事项（必写）
  • • robots.txt 解读
    • 频率控制的数学原理
    • 合规红线（重要）
    • 伦理准则
  • 4️⃣ 技术选型与整体流程（What/How）
  • • 静态 vs 动态 vs API 的判断方法
    • 技术选型对比表
    • 整体流程图
  • 5️⃣ 环境准备与依赖安装（可复现）
  • • Python 版本要求
    • 依赖清单与作用说明
    • 项目目录结构（生产级别）
  • 6️⃣ 核心实现：请求层（Fetcher）
  • • 完整代码实现
    • 代码解析：为什么这样写？
    • • 1. 为什么使用 `requests.Session()`？
      • 2. `@retry` 装饰器的工作原理
      • 3. 频率限制的数学模型
  • 7️⃣ 核心实现：解析层（Parser）
  • • HTML结构分析（关键步骤）
    • 完整代码实现
    • 代码解析：选择器定位技巧
    • • 1. CSS Selector vs XPath 的选择
      • 2. 为什么要用 `select_one()` 而不是 `find()`？
      • 3. 容错处理的三重保险
    • 正则表达式的妙用
    • • 案例1：提取评分
      • 案例2：清除前缀后缀
  • 8️⃣ 数据清洗与转换（Cleaner）
  • • 完整代码实现
    • 代码解析：数字转换算法
    • • 难点1：中文数量词的转换
      • 难点2：处理"+"号
      • 难点3：小数点的处理
  • 9️⃣ 数据存储与导出（Storage）
  • • 完整代码实现
    • 代码解析：数据去重策略
    • • 为什么要去重？
      • `keep='first'` vs `keep='last'` 的选择
  • 🔟 运行方式与结果展示（必写）
  • • 主程序整合
    • 启动命令
    • 输出结构
    • 示例输出（CSV格式）
    • 示例输出（JSON格式）
    • 控制台输出（日志）
  • 1️⃣1️⃣ 常见问题与排错（强烈建议写）
  • • Q1: 403 Forbidden – 被识别为爬虫
    • Q2: 429 Too Many Requests – 触发频率限制
    • Q3: 页面结构变化导致解析失败
    • Q4: 编码乱码问题
    • Q5: 内存溢出（爬取大量数据时）
    • Q6: 动态内容未加载
  • 1️⃣2️⃣ 进阶优化（可选但加分）
  • • 并发爬取：提升10倍效率
    • • 方案1：使用 `concurrent.futures`
      • 方案2：使用 `asyncio` + `aiohttp`
    • 断点续跑：避免重复劳动
    • 增量更新：只爬新数据
    • 监控与告警：自动检测异常
    • 定时任务：自动化运行
  • 1️⃣3️⃣ 总结与延伸阅读
  • • 我们完成了什么？
    • 这个项目的真实价值
    • 还有哪些可以优化的地方？
    • 延伸阅读推荐
  • 🌟 文末
  • • 📌 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集

🌟 开篇语

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

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

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

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

📣 专栏推广时间：如果你想系统学爬虫，而不是碎片化东拼西凑，欢迎订阅/关注专栏《Python爬虫实战》 订阅后更新会优先推送，按目录学习更高效～

1️⃣ 摘要（Abstract）

一句话概括：使用 Python + requests + BeautifulSoup 爬取Chrome Web Store扩展商店的分类页面，提取插件名称、评分、下载量、详情链接等关键信息，最终输出为CSV和JSON格式的结构化数据。

读完你能获得：

• 深入理解静态HTML页面的爬取流程与选择器定位技巧
• 掌握如何处理分页数据、动态加载内容的识别与应对策略
• 学会构建健壮的数据清洗管道，处理缺失值、格式化数字、统一编码
• 理解反爬虫对抗的基本原理：User-Agent伪装、请求频率控制、异常重试机制

2️⃣ 背景与需求（Why）

为什么要爬取插件市场数据？

作为产品经理、开发者或者数据分析师，你可能面临以下场景：

场景一：市场竞品分析 你正在开发一款浏览器插件，想知道同类产品的用户评分、下载量、用户反馈趋势，从而优化自己的产品定位。通过批量爬取分类页的插件数据，你可以快速构建竞品数据库。

场景二：用户需求挖掘 某个细分领域（比如"开发者工具"）有哪些高分插件？这些插件解决了什么痛点？通过数据聚合，你能发现未被满足的用户需求。

场景三：学术研究 研究浏览器扩展生态的演化规律、用户行为模式、开发者社区活跃度等课题，需要大量真实的插件元数据作为研究样本。

目标站点与字段清单

目标平台：Chrome Web Store（https://chromewebstore.google.com/）

核心数据字段：

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

robots.txt 解读

访问 https://chromewebstore.google.com/robots.txt 可以看到爬虫协议：

User–agent: *
Disallow: /search
Allow: /

解读：

• ✅ 允许抓取：分类页、详情页等公开页面
• ❌ 禁止抓取：搜索结果页（/search路径）

频率控制的数学原理

假设Chrome Web Store每秒能处理1000个请求，而你的爬虫每秒发送100个请求，虽然服务器能承受，但会被识别为异常流量。

黄金法则：模拟真实用户行为

• 人类浏览一个页面平均需要 2-5秒
• 翻页操作间隔 3-8秒
• 点击详情页间隔 1-3秒

# ❌ 错误示范：瞬间发送1000个请求
for url in urls:
requests.get(url) # 0.1秒内全部发完，100%被封

# ✅ 正确示范：随机延迟+指数退避
import time
import random

for url in urls:
response = requests.get(url)
# 基础延迟：2-5秒
base_delay = random.uniform(2, 5)
# 如果触发限流（429），指数级增加等待时间
if response.status_code == 429:
base_delay *= 3
time.sleep(base_delay)

合规红线（重要）

✅ 允许的操作：

• 抓取公开展示的插件名称、评分、下载量等元数据
• 用于个人学习、数据分析、学术研究
• 遵守 robots.txt 规则，控制请求频率

❌ 禁止的操作：

• 批量下载插件安装包（侵犯版权）
• 绕过登录限制抓取用户评论、个人信息（侵犯隐私）
• 使用数据进行商业牟利而不经授权
• 高频攻击式爬取导致服务器过载

法律依据：

• 《中华人民共和国网络安全法》第44条：不得危害网络安全
• 美国《计算机欺诈与滥用法》（CFAA）：禁止未经授权访问

伦理准则

我写这篇教程的目的是技术交流与教育，而非鼓励任何违法行为。请在使用时：

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

静态 vs 动态 vs API 的判断方法

打开Chrome扩展商店的分类页面（例如"生产工具"分类），按下 F12 打开开发者工具：

步骤一：查看Network标签

步骤二：查看Elements标签

判断结果：Chrome Web Store的分类页属于静态HTML + 部分动态加载的混合模式：

• 首屏内容是静态HTML
• 滚动加载更多内容时，会发送API请求

我们的策略：

• 先用 requests + BeautifulSoup 抓取首屏数据（快速验证）
• 如果需要深度翻页，再分析API接口或使用Selenium

技术选型对比表

本项目选择：requests + BeautifulSoup + lxml 组合

理由：

整体流程图

┌─────────────────────┐
│ 输入：分类页URL │
│ (如：生产工具分类) │
└──────────┬──────────┘
│
▼
┌─────────────────────┐
│ 请求层 (Fetcher) │
│ • 构造headers │ ← User–Agent伪装
│ • 设置timeout │ ← 防止卡死
│ • 发送GET请求 │
└──────────┬──────────┘
│
▼
┌─────────────────────┐
│ 解析层 (Parser) │
│ • 定位插件卡片容器 │ ← CSS Selector / XPath
│ • 循环提取每个插件 │
│ • 处理评分文本 │ ← "4.5(1.2万)"→4.5+12000
│ • 处理下载量文本 │ ← "100万+"→1000000
└──────────┬──────────┘
│
▼
┌─────────────────────┐
│ 清洗层 (Cleaner) │
│ • 去除空值/异常值 │
│ • 数字格式标准化 │ ← "1.2万"→12000
│ • URL完整性校验 │
└──────────┬──────────┘
│
▼
┌─────────────────────┐
│ 存储层 (Storage) │
│ • 去重（基于ID） │
│ • 保存为CSV │ ← pandas.to_csv
│ • 保存为JSON │ ← json.dump
└─────────────────────┘

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

Python 版本要求

# 检查Python版本
python –version # 需要 >= 3.8

# 如果版本过低，建议使用pyenv管理多版本
pyenv install 3.10.0
pyenv local 3.10.0

依赖清单与作用说明

创建 requirements.txt：

requests==2.31.0 # HTTP请求库，核心工具
beautifulsoup4==4.12.2 # HTML解析库
lxml==4.9.3 # BS4的解析器，速度比html.parser快
pandas==2.1.0 # 数据处理与CSV导出
fake-useragent==1.2.1 # 随机生成User-Agent
retrying==1.3.4 # 自动重试装饰器
loguru==0.7.0 # 更优雅的日志库

安装命令：

# 方式一：直接安装
pip install requests beautifulsoup4 lxml pandas fake-useragent retrying loguru

# 方式二：从requirements.txt安装（推荐）
pip install -r requirements.txt

# 方式三：使用清华镜像源加速（国内用户）
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

项目目录结构（生产级别）

chrome_plugin_crawler/
│
├── main.py # 主程序入口
├── config.py # 配置文件（URL、参数等）
├── requirements.txt # 依赖清单
├── README.md # 项目说明文档
│
├── core/ # 核心功能模块
│ ├── __init__.py
│ ├── fetcher.py # 请求层：负责HTTP请求
│ ├── parser.py # 解析层：提取数据
│ ├── cleaner.py # 清洗层：数据规范化
│ └── storage.py # 存储层：保存数据
│
├── utils/ # 工具函数
│ ├── __init__.py
│ ├── logger.py # 日志配置
│ ├── retry.py # 重试装饰器
│ └── text_utils.py # 文本处理工具
│
├── data/ # 数据输出目录
│ ├── raw/ # 原始HTML（调试用）
│ ├── csv/ # CSV格式
│ └── json/ # JSON格式
│
├── logs/ # 日志文件
│ └── crawler.log
│
└── tests/ # 单元测试（可选）
├── test_parser.py
└── test_cleaner.py

创建目录的脚本：

# create_structure.sh
mkdir -p chrome_plugin_crawler/{core,utils,data/{raw,csv,json},logs,tests}
touch chrome_plugin_crawler/core/__init__.py
touch chrome_plugin_crawler/utils/__init__.py

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

请求层是整个爬虫的第一道防线，需要处理网络异常、反爬检测、请求重试等问题。

完整代码实现

# core/fetcher.py
import requests
import time
import random
from typing import Optional
from fake_useragent import UserAgent
from retrying import retry
from loguru import logger

class ChromeStoreFetcher:
"""Chrome扩展商店请求器

职责：
1. 发送HTTP请求获取HTML页面
2. 处理反爬策略（headers伪装、频率控制）
3. 异常处理与自动重试
4. Session管理（保持连接复用）
"""

def __init__(self, proxy: Optional[dict] = None):
"""
初始化请求器

Args:
proxy: 代理配置，格式 {'http': 'http://ip:port', 'https': 'https://ip:port'}
"""
# 创建Session对象，复用TCP连接，提升性能
self.session = requests.Session()

# User-Agent池，随机选择避免被识别
self.ua = UserAgent()

# 代理设置（如果提供）
self.proxy = proxy

# 请求计数器，用于统计和限流
self.request_count = 0

# 上次请求时间，用于控制频率
self.last_request_time = 0

logger.info("ChromeStoreFetcher 初始化完成")

def _get_headers(self) –> dict:
"""
生成请求头

为什么需要这么多header字段？
• User-Agent: 标识浏览器类型，必须伪装
• Accept: 告诉服务器我们接受的内容类型
• Accept-Language: 语言偏好，影响返回内容
• Accept-Encoding: 支持压缩，减少传输量
• Referer: 来源页面，模拟真实跳转
• Connection: keep-alive 保持连接复用
"""
return {
'User-Agent': self.ua.random, # 每次请求随机UA
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Accept-Encoding': 'gzip, deflate, br',
'Referer': 'https://chromewebstore.google.com/',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
# Chrome特有的请求头，增加真实性
'Sec-Fetch-Dest': 'document',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'same-origin',
'Cache-Control': 'max-age=0',
}

def _rate_limit(self):
"""
请求频率限制

原理：确保两次请求之间至少间隔 min_interval 秒

为什么用指数退避？
• 请求次数越多，被检测的概率越高
• 适当增加延迟，降低被封风险
"""
min_interval = 2 # 基础间隔2秒

# 指数退避：每100次请求，间隔增加1秒
extra_delay = (self.request_count // 100) * 1
total_interval = min_interval + extra_delay

# 计算距离上次请求过了多久
elapsed = time.time() – self.last_request_time

# 如果间隔不够，就等待
if elapsed < total_interval:
sleep_time = total_interval – elapsed
# 加入随机波动，更像人类行为
sleep_time += random.uniform(0, 1)
logger.debug(f"频率限制：等待 {sleep_time:.2f} 秒")
time.sleep(sleep_time)

@retry(
stop_max_attempt_number=3, # 最多重试3次
wait_exponential_multiplier=1000, # 等待时间指数增长：1s, 2s, 4s
wait_exponential_max=10000 # 最大等待10秒
)
def fetch(self, url: str, params: Optional[dict] = None) –> Optional[requests.Response]:
"""
发送GET请求

Args:
url: 目标URL
params: URL参数字典

Returns:
Response对象，失败返回None

异常处理：
• Timeout: 请求超时
• ConnectionError: 网络连接失败
• HTTPError: HTTP错误状态码（4xx, 5xx）
• RequestException: requests库的通用异常
"""
# 频率控制
self._rate_limit()

try:
# 发送请求
response = self.session.get(
url,
params=params,
headers=self._get_headers(),
proxies=self.proxy,
timeout=15, # 15秒超时
allow_redirects=True # 允许重定向
)

# 更新统计信息
self.request_count += 1
self.last_request_time = time.time()

# 检查状态码
if response.status_code == 200:
logger.info(f"✓ 请求成功: {url[:60]}… [状态码: 200]")
return response

elif response.status_code == 429:
# 被限流，打印警告但让重试机制处理
logger.warning(f"⚠ 触发限流 (429)，即将重试…")
raise requests.HTTPError("Rate limited")

elif response.status_code == 403:
logger.error(f"✗ 被拒绝访问 (403)：可能被识别为爬虫")
# 403通常不是临时问题，不重试
return None

else:
logger.error(f"✗ 请求失败: 状态码 {response.status_code}")
response.raise_for_status() # 触发HTTPError，让重试机制处理

except requests.Timeout:
logger.warning(f"⏱ 请求超时: {url[:60]}…")
raise # 让重试装饰器处理

except requests.ConnectionError as e:
logger.error(f"🔌 网络连接失败: {str(e)[:100]}")
raise

except requests.HTTPError as e:
logger.error(f"📡 HTTP错误: {str(e)}")
raise

except Exception as e:
logger.error(f"💥 未知异常: {type(e).__name__} – {str(e)[:100]}")
return None

def save_html(self, html: str, filename: str):
"""
保存HTML到本地（调试用）

Args:
html: HTML内容
filename: 文件名
"""
filepath = f"data/raw/{filename}.html"
try:
with open(filepath, 'w', encoding='utf-8') as f:
f.write(html)
logger.debug(f"HTML已保存: {filepath}")
except Exception as e:
logger.error(f"保存HTML失败: {str(e)}")

def close(self):
"""关闭Session，释放资源"""
self.session.close()
logger.info(f"Session已关闭，共发送 {self.request_count} 次请求")

# 使用示例
if __name__ == "__main__":
fetcher = ChromeStoreFetcher()

# 测试请求
url = "https://chromewebstore.google.com/category/extensions/productivity"
response = fetcher.fetch(url)

if response:
print(f"页面大小: {len(response.text)} 字符")
fetcher.save_html(response.text, "test_page")

fetcher.close()

代码解析：为什么这样写？

1. 为什么使用 requests.Session()？

# ❌ 不推荐：每次都创建新连接
for url in urls:
requests.get(url) # 每次都要经历：DNS解析 → TCP握手 → TLS握手

# ✅ 推荐：复用连接
session = requests.Session()
for url in urls:
session.get(url) # 复用已建立的连接，速度快3-5倍

性能对比：

• 不使用Session：100个请求耗时 ~30秒
• 使用Session：100个请求耗时 ~10秒

2. @retry 装饰器的工作原理

@retry(
stop_max_attempt_number=3, # 最多3次
wait_exponential_multiplier=1000, # 1秒基数
wait_exponential_max=10000 # 最大10秒
)
def fetch(...):
...

重试时间线：

• 第1次失败：等待 2^1 × 1秒 = 2秒
• 第2次失败：等待 2^2 × 1秒 = 4秒
• 第3次失败：等待 min(2^3 × 1秒, 10秒) = 8秒

为什么用指数退避？

• 如果服务器暂时过载，等待时间越长，恢复的可能性越大
• 避免"雪崩效应"：大量失败请求立即重试，导致服务器崩溃

3. 频率限制的数学模型

def _rate_limit(self):
min_interval = 2
extra_delay = (self.request_count // 100) * 1
total_interval = min_interval + extra_delay

请求次数与间隔关系：

• 前100次请求：间隔 2秒
• 101-200次：间隔 3秒
• 201-300次：间隔 4秒
• …以此类推

为什么这样设计？

• 初期快速采集，提高效率
• 后期减速，降低被检测风险
• 模拟人类行为：刚开始兴奋，越看越慢

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

解析层负责从HTML中提取结构化数据，这是爬虫的核心逻辑。

HTML结构分析（关键步骤）

在编写解析代码之前，必须先分析目标网页的HTML结构。

步骤一：打开Chrome扩展商店的分类页

https://chromewebstore.google.com/category/extensions/productivity

步骤二：右键点击某个插件卡片 → “检查元素”

你会看到类似这样的HTML结构（简化版）：

<div class="plugin-card" data-id="nmmhkkegccagdldgiimedpiccmgmieda">
<!– 插件图标 –>
<img src="…icon.png" alt="AdBlock">

<!– 插件信息容器 –>
<div class="plugin-info">
<!– 插件名称 –>
<h3 class="plugin-title">
<a href="/detail/nmmhkkegccagdldgiimedpiccmgmieda">
AdBlock — 最佳广告拦截工具
</a>
</h3>

<!– 开发者 –>
<div class="developer">
由 AdBlock Inc. 提供
</div>

<!– 评分 –>
<div class="rating">
<span class="stars" aria-label="评分 4.6 星，满分 5 星">★★★★☆</span>
<span class="rating-value">4.6</span>
<span class="rating-count">(12.5万)</span>
</div>

<!– 用户数 –>
<div class="users">
1000万+ 位用户
</div>

<!– 描述 –>
<p class="description">
阻止YouTube、Facebook等网站上的烦人广告…
</p>
</div>
</div>

步骤三：确定CSS选择器

完整代码实现

# core/parser.py
from bs4 import BeautifulSoup
from typing import List, Dict, Optional
import re
from loguru import logger

class ChromeStoreParser:
"""Chrome扩展商店HTML解析器

职责：
1. 定位插件卡片的DOM元素
2. 提取各个字段的原始值
3. 初步清洗数据（去除空格、换行等）

注意：复杂的数字转换由 Cleaner 模块负责
"""

def __init__(self):
self.parsed_count = 0

def parse_category_page(self, html: str) –> List[Dict]:
"""
解析分类页面，提取所有插件信息

Args:
html: 页面HTML字符串

Returns:
插件信息列表，每个元素是一个字典
"""
# 使用lxml解析器（速度快）
soup = BeautifulSoup(html, 'lxml')

# 定位所有插件卡片
# 注意：实际的class名可能不同，需要根据真实页面调整
plugin_cards = soup.select('div[role="listitem"]') # 真实选择器

if not plugin_cards:
logger.warning("⚠ 未找到插件卡片，页面结构可能已变化")
return []

logger.info(f"找到 {len(plugin_cards)} 个插件卡片")

plugins = []
for index, card in enumerate(plugin_cards, 1):
try:
plugin_data = self._parse_single_plugin(card, index)
if plugin_data:
plugins.append(plugin_data)
self.parsed_count += 1
except Exception as e:
logger.error(f"解析第 {index} 个插件失败: {str(e)}")
continue

logger.info(f"✓ 成功解析 {len(plugins)} 个插件")
return plugins

def _parse_single_plugin(self, card: BeautifulSoup, index: int) –> Optional[Dict]:
"""
解析单个插件卡片

Args:
card: 插件卡片的BeautifulSoup对象
index: 序号（用于日志）

Returns:
插件信息字典，失败返回None
"""
plugin = {}

# 1. 提取插件名称和链接
title_elem = card.select_one('h3 a, div[role="heading"] a')
if title_elem:
plugin['plugin_name'] = self._clean_text(title_elem.get_text())
plugin['detail_url'] = self._build_full_url(title_elem.get('href', ''))
plugin['plugin_id'] = self._extract_id_from_url(plugin['detail_url'])
else:
logger.warning(f"插件 #{index}: 未找到标题元素")
return None

# 2. 提取开发者
developer_elem = card.select_one('div[data-tooltip], span.developer')
if developer_elem:
developer_text = self._clean_text(developer_elem.get_text())
# 去除前缀"由…提供"
plugin['developer'] = re.sub(r'^由\\s*|提供$', '', developer_text)
else:
plugin['developer'] = ''

# 3. 提取评分
rating_elem = card.select_one('span[aria-label*="评分"]')
if rating_elem:
# 从 aria-label 中提取：「评分 4.6 星，满分 5 星」→ 4.6
aria_label = rating_elem.get('aria-label', '')
rating_match = re.search(r'(\\d+\\.?\\d*)\\s*星', aria_label)
if rating_match:
plugin['rating'] = rating_match.group(1)
else:
plugin['rating'] = ''
else:
plugin['rating'] = ''

# 4. 提取评分人数
rating_count_elem = card.select_one('span.rating-count, span:contains("评分")')
if rating_count_elem:
# 原始文本："(12.5万)" 或 "12.5万 个评分"
text = self._clean_text(rating_count_elem.get_text())
plugin['rating_count_text'] = text
else:
plugin['rating_count_text'] = ''

# 5. 提取用户数量
users_elem = card.select_one('span.users, span:contains("用户")')
if users_elem:
# 原始文本："1000万+ 位用户"
text = self._clean_text(users_elem.get_text())
plugin['users_count_text'] = text
else:
plugin['users_count_text'] = ''

# 6. 提取描述
desc_elem = card.select_one('div.description, p')
if desc_elem:
plugin['description'] = self._clean_text(desc_elem.get_text())
# 限制长度，避免过长
if len(plugin['description']) > 200:
plugin['description'] = plugin['description'][:200] + '…'
else:
plugin['description'] = ''

# 7. 提取分类（如果页面包含）
category_elem = card.select_one('span.category')
if category_elem:
plugin['category'] = self._clean_text(category_elem.get_text())
else:
plugin['category'] = '' # 可以在外部补充

return plugin

def _clean_text(self, text: str) –> str:
"""
清理文本：去除多余空白字符

Args:
text: 原始文本

Returns:
清理后的文本
"""
if not text:
return ''
# 去除首尾空白
text = text.strip()
# 合并多个空格为一个
text = re.sub(r'\\s+', ' ', text)
return text

def _build_full_url(self, relative_url: str) –> str:
"""
构建完整URL

Args:
relative_url: 相对路径，如 "/detail/abc123"

Returns:
完整URL
"""
if not relative_url:
return ''

base_url = 'https://chromewebstore.google.com'

if relative_url.startswith('http'):
return relative_url
elif relative_url.startswith('/'):
return base_url + relative_url
else:
return base_url + '/' + relative_url

def _extract_id_from_url(self, url: str) –> str:
"""
从URL中提取插件ID

Args:
url: 详情页URL，如 "https://…/detail/abc123def456"

Returns:
插件ID，如 "abc123def456"
"""
if not url:
return ''

# 匹配 /detail/ 后面的字符串
match = re.search(r'/detail/([a-z0-9]+)', url)
if match:
return match.group(1)

# 兜底：返回URL最后一段
return url.split('/')[–1]

# 使用示例
if __name__ == "__main__":
# 读取测试HTML
with open('data/raw/test_page.html', 'r', encoding='utf-8') as f:
html = f.read()

parser = ChromeStoreParser()
plugins = parser.parse_category_page(html)

# 打印前3个结果
for i, plugin in enumerate(plugins[:3], 1):
print(f"\\n=== 插件 {i} ===")
for key, value in plugin.items():
print(f"{key}: {value}")

代码解析：选择器定位技巧

1. CSS Selector vs XPath 的选择

CSS Selector（本项目使用）：

# 优点：语法简洁，性能好
soup.select('div.plugin-card') # 选择所有class为plugin-card的div
soup.select_one('h3 a') # 选择h3下的第一个a标签

XPath：

# 优点：功能强大，可以逆向查找父元素
soup.find_all('//div[@class="plugin-card"]') # lxml支持

选择建议：

• 简单场景用CSS Selector（90%的情况）
• 需要复杂逻辑（如"找到包含某文本的div的父元素"）用XPath

2. 为什么要用 select_one() 而不是 find()？

# select_one: 返回第一个匹配元素，不匹配返回None
elem = soup.select_one('h3 a')
if elem:
text = elem.get_text() # 安全

# find: 功能相同，但语法稍冗长
elem = soup.find('a', class_='title')

结论：两者功能相同，select_one() 更符合CSS思维。

3. 容错处理的三重保险

# 第一重：使用 .get() 避免KeyError
href = elem.get('href', '') # 如果没有href属性，返回空字符串

# 第二重：使用 if elem 避免NoneType错误
if elem:
text = elem.get_text()

# 第三重：使用 try-except 捕获异常
try:
plugin_data = self._parse_single_plugin(card)
except Exception as e:
logger.error(f"解析失败: {e}")
continue

正则表达式的妙用

案例1：提取评分

# 输入：「评分 4.6 星，满分 5 星」
# 目标：提取 4.6

aria_label = "评分 4.6 星，满分 5 星"
rating_match = re.search(r'(\\d+\\.?\\d*)\\s*星', aria_label)
# 解释：
# \\d+ 匹配一个或多个数字
# \\.? 匹配0个或1个小数点
# \\d* 匹配0个或多个数字
# \\s* 匹配0个或多个空白符
# 星 匹配汉字"星"

if rating_match:
rating = float(rating_match.group(1)) # 4.6

案例2：清除前缀后缀

# 输入："由 AdBlock Inc. 提供"
# 目标："AdBlock Inc."

developer_text = "由 AdBlock Inc. 提供"
clean = re.sub(r'^由\\s*|提供$', '', developer_text)
# 解释：
# ^由\\s* 匹配开头的"由"和后面的空格
# | 或
# 提供$ 匹配结尾的"提供"
# 替换为空字符串

print(clean) # "AdBlock Inc."

8️⃣ 数据清洗与转换（Cleaner）

解析层提取的是原始文本，需要进一步转换为标准格式。

完整代码实现

# core/cleaner.py
import re
from typing import Dict, List, Any
from loguru import logger

class DataCleaner:
"""数据清洗器

职责：
1. 将文本数字转为int/float（如"12.5万"→125000）
2. 去除异常值和缺失值
3. 统一数据格式
"""

def __init__(self):
# 中文数字单位映射
self.unit_map = {
'万': 10000,
'亿': 100000000,
'k': 1000,
'K': 1000,
'm': 1000000,
'M': 1000000,
'w': 10000, # 部分网站用w表示万
}

def clean_plugins(self, plugins: List[Dict]) –> List[Dict]:
"""
批量清洗插件数据

Args:
plugins: 原始插件列表

Returns:
清洗后的插件列表
"""
cleaned = []
for plugin in plugins:
try:
cleaned_plugin = self.clean_single_plugin(plugin):
logger.error(f"清洗插件 {plugin.get('plugin_name', 'Unknown')} 失败: {e}")
continue

logger.info(f"✓ 清洗完成: {len(cleaned)}/{len(plugins)} 个插件有效")
return cleaned

def clean_single_plugin(self, plugin: Dict) –> Dict:
"""清洗单个插件数据"""
cleaned = plugin.copy()

# 1. 清洗评分（字符串 → float）
cleaned['rating'] = self._parse_rating(plugin.get('rating', ''))

# 2. 清洗评分人数（"12.5万" → 125000）
cleaned['rating_count'] = self._parse_count(plugin.get('rating_count_text', ''))

# 3. 清洗用户数量（"1000万+" → 10000000）
cleaned['users_count'] = self._parse_count(plugin.get('users_count_text', ''))

# 4. 验证必填字段
if not cleaned.get('plugin_id'):
logger.warning(f"插件缺少ID: {plugin.get('plugin_name')}")
return None

if not cleaned.get('plugin_name'):
logger.warning(f"插件缺少名称: {plugin.get('plugin_id')}")
return None

return cleaned

def _parse_rating(self, rating_text: str) –> float:
"""
解析评分

Args:
rating_text: "4.6" 或 "4.65"

Returns:
float格式的评分，失败返回0.0
"""
if not rating_text:
return 0.0

try:
# 去除空格和特殊字符，只保留数字和小数点
clean_text = re.sub(r'[^\\d.]', '', str(rating_text))
rating = float(clean_text)

# 验证范围（评分通常是0-5）
if 0 <= rating <= 5:
return round(rating, 2)
else:
logger.warning(f"评分超出范围: {rating}")
return 0.0
except ValueError:
logger.warning(f"评分格式错误: {rating_text}")
return 0.0

def _parse_count(self, count_text: str) –> int:
"""
解析数量（支持中文单位）

Args:
count_text:
"12.5万" → 125000
"1000万+" → 10000000
"1.5K" → 1500
"50" → 50

Returns:
int格式的数量，失败返回0
"""
if not count_text:
return 0

# 去除"位用户"、"个评分"、"+"等无关字符
clean_text = re.sub(r'[位个用户评分\\+\\s]', '', count_text)

try:
# 提取数字和单位
# 正则：(\\d+\\.?\\d*) 匹配数字，([万亿kKmMw])? 匹配单位
match = re.search(r'(\\d+\\.?\\d*)([万亿kKmMw])?', clean_text)

if not match:
return 0

number = float(match.group(1))
unit = match.group(2)

# 根据单位进行换算
if unit and unit in self.unit_map:
number *= self.unit_map[unit]

return int(number)

except (ValueError, AttributeError) as e:
logger.warning(f"数量格式错误: {count_text} – {e}")
return 0

def validate_url(self, url: str) –> bool:
"""验证URL是否有效"""
if not url:
return False
return url.startswith('http')

# 单元测试
if __name__ == "__main__":
cleaner = DataCleaner()

# 测试评分解析
assert cleaner._parse_rating("4.6") == 4.6
assert cleaner._parse_rating("4.65") == 4.65
assert cleaner._parse_rating("") == 0.0

# 测试数量解析
assert cleaner._parse_count("12.5万") == 125000
assert cleaner._parse_count("1000万+") == 10000000
assert cleaner._parse_count("1.5K") == 1500
assert cleaner._parse_count("50") == 50
assert cleaner._parse_count("1.2亿") == 120000000

print("✓ 所有测试通过！")

代码解析：数字转换算法

难点1：中文数量词的转换

# 输入："12.5万"
# 步骤：
# 1. 正则提取：12.5 + 万
# 2. 查表：万 = 10000
# 3. 计算：12.5 × 10000 = 125000

match = re.search(r'(\\d+\\.?\\d*)([万亿kKmMw])?', "12.5万")
number = float(match.group(1)) # 12.5
unit = match.group(2) # 万
result = int(number * 10000) # 125000

难点2：处理"+"号

# 输入："1000万+"
# 问题："+"不能参与数学运算
# 解决方案：用正则去除所有非数字字符（除了小数点和单位）

clean_text = re.sub(r'[位个用户评分\\+\\s]', '', "1000万+ 位用户")
# 结果："1000万"

难点3：小数点的处理

# 输入："1.5万"
# 陷阱：不能直接 int("1.5万")
# 正确流程：
# 1. 提取：1.5 + 万
# 2. 计算：1.5 × 10000 = 15000.0
# 3. 转int：int(15000.0) = 15000

number = 1.5
unit_value = 10000
result = int(number * unit_value)

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

存储层负责将清洗后的数据持久化到文件系统。

完整代码实现

# core/storage.py
import json
import pandas as pd
from pathlib import Path
from typing import List, Dict
from datetime import datetime
from loguru import logger

class DataStorage:
"""数据存储管理器

职责：
1. 保存数据为CSV格式（Excel友好）
2. 保存数据为JSON格式（保留完整结构）
3. 数据去重（基于plugin_id）
4. 生成数据统计报告
"""

def __init__(self, output_dir: str = "data"):
self.output_dir = Path(output_dir)
self.csv_dir = self.output_dir / "csv"
self.json_dir = self.output_dir / "json"

# 创建目录
self.csv_dir.mkdir(parents=True, exist_ok=True)
self.json_dir.mkdir(parents=True, exist_ok=True)

logger.info(f"存储目录已初始化: {self.output_dir}")

def save_csv(self, plugins: List[Dict], filename: str = None):
"""
保存为CSV格式

Args:
plugins: 插件数据列表
filename: 文件名（不含扩展名），默认使用时间戳
"""
if not plugins:
logger.warning("没有数据可保存")
return

# 生成文件名
if not filename:
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"plugins_{timestamp}"

filepath = self.csv_dir / f"{filename}.csv"

try:
# 转DataFrame
df = pd.DataFrame(plugins)

# 去重
before_count = len(df)
df = df.drop_duplicates(subset=['plugin_id'], keep='first')
after_count = len(df)

if before_count > after_count:
logger.info(f"去重：移除 {before_count – after_count} 条重复数据")

# 字段排序（方便阅读）
columns_order = [
'plugin_id', 'plugin_name', 'category',
'rating', 'rating_count', 'users_count',
'developer', 'detail_url', 'description'
]
# 保留原有字段 + 排序
existing_cols = [col for col in columns_order if col in df.columns]
other_cols = [col for col in df.columns if col not in columns_order]
df = df[existing_cols + other_cols]

# 保存CSV（使用utf-8-sig避免Excel乱码）
df.to_csv(filepath, index=False, encoding='utf-8-sig')

logger.info(f"✓ CSV已保存: {filepath} ({len(df)} 行)")

# 打印统计信息
self._print_statistics(df)

except Exception as e:
logger.error(f"保存CSV失败: {str(e)}")

def save_json(self, plugins: List[Dict], filename: str = None):
"""
保存为JSON格式

Args:
plugins: 插件数据列表
filename: 文件名（不含扩展名）
"""
if not plugins:
logger.warning("没有数据可保存")
return

if not filename:
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"plugins_{timestamp}"

filepath = self.json_dir / f"{filename}.json"

try:
# 添加元数据
output = {
'metadata': {
'total_count': len(plugins),
'export_time': datetime.now().isoformat(),
'version': '1.0'
},
'data': plugins
}

# 保存JSON
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(output, f, ensure_ascii=False, indent=2)

logger.info(f"✓ JSON已保存: {filepath} ({len(plugins)} 条)")

except Exception as e:
logger.error(f"保存JSON失败: {str(e)}")

def _print_statistics(self, df: pd.DataFrame):
"""打印数据统计"""
try:
logger.info("=" * 50)
logger.info("📊 数据统计报告")
logger.info("=" * 50)

# 基本统计
logger.info(f" • 总插件数: {len(df)}")
logger.info(f" • 分类数: {df['category'].nunique() if 'category' in df else 'N/A'}")

# 评分统计
if 'rating' in df and df['rating'].sum() > 0:
avg_rating = df[df['rating'] > 0]['rating'].mean()
logger.info(f" • 平均评分: {avg_rating:.2f} 星")

# 评分分布
high_rated = len(df[df['rating'] >= 4.0])
logger.info(f" • 高分插件(≥4.0): {high_rated} 个 ({high_rated/len(df)*100:.1f}%)")

# 用户数统计
if 'users_count' in df and df['users_count'].sum() > 0:
total_users = df['users_count'].sum()
logger.info(f" • 总用户数: {total_users:,}")

# Top 5插件
top5 = df.nlargest(5, 'users_count')[['plugin_name', 'users_count']]
logger.info("\\n 🏆 用户数Top 5:")
for idx, row in top5.iterrows():
logger.info(f" {row['plugin_name'][:30]}: {row['users_count']:,} 用户")

logger.info("=" * 50)

except Exception as e:
logger.warning(f"统计信息生成失败: {e}")

def load_csv(self, filename: str) –> pd.DataFrame:
"""加载CSV文件"""
filepath = self.csv_dir / f"{filename}.csv"
try:
df = pd.read_csv(filepath, encoding='utf-8-sig')
logger.info(f"✓ 已加载: {filepath} ({len(df)} 行)")
return df
except FileNotFoundError:
logger.error(f"文件不存在: {filepath}")
return pd.DataFrame()

def merge_csv_files(self, pattern: str = "plugins_*.csv"):
"""
合并多个CSV文件（用于批量爬取）

Args:
pattern: 文件匹配模式
"""
csv_files = list(self.csv_dir.glob(pattern))

if not csv_files:
logger.warning(f"未找到匹配的文件: {pattern}")
return

logger.info(f"找到 {len(csv_files)} 个文件，开始合并…")

dfs = []
for file in csv_files:
df = pd.read_csv(file, encoding='utf-8-sig')
dfs.append(df)

# 合并并去重
merged = pd.concat(dfs, ignore_index=True)
merged = merged.drop_duplicates(subset=['plugin_id'], keep='first')

# 保存
output_file = self.csv_dir / "merged_plugins.csv"
merged.to_csv(output_file, index=False, encoding='utf-8-sig')

logger.info(f"✓ 合并完成: {output_file} ({len(merged)} 行)")

# 使用示例
if __name__ == "__main__":
storage = DataStorage()

# 模拟数据
test_plugins = [
{
'plugin_id': 'abc123',
'plugin_name': 'Test Plugin 1',
'rating': 4.5,
'users_count': 100000
},
{
'plugin_id': 'def456',
'plugin_name': 'Test Plugin 2',
'rating': 4.8,
'users_count': 500000
}
]

storage.save_csv(test_plugins, 'test_output')
storage.save_json(test_plugins, 'test_output')

代码解析：数据去重策略

为什么要去重？

# 场景：多次爬取同一分类，可能抓到重复数据
plugins = [
{'plugin_id': 'abc', 'name': 'AdBlock'},
{'plugin_id': 'abc', 'name': 'AdBlock'}, # 重复！
{'plugin_id': 'def', 'name': 'Grammarly'}
]

# 去重后
df = pd.DataFrame(plugins)
df = df.drop_duplicates(subset=['plugin_id'], keep='first')
# 结果：只保留2行

keep='first' vs keep='last' 的选择

# keep='first': 保留第一次出现的记录
df.drop_duplicates(subset=['plugin_id'], keep='first')
# 适用场景：首次爬取的数据更可靠

# keep='last': 保留最后一次出现的记录
df.drop_duplicates(subset=['plugin_id'], keep='last')
# 适用场景：增量更新，保留最新数据

# keep=False: 删除所有重复记录
df.drop_duplicates(subset=['plugin_id'], keep=False)
# 适用场景：只保留唯一数据，丢弃所有重复项

🔟 运行方式与结果展示（必写）

主程序整合

# main.py
from core.fetcher import ChromeStoreFetcher
from core.parser import ChromeStoreParser
from core.cleaner import DataCleaner
from core.storage import DataStorage
from loguru import logger
import sys

# 配置日志
logger.remove() # 移除默认handler
logger.add(
sys.stdout,
format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | <level>{message}</level>",
level="INFO"
)
logger.add(
"logs/crawler_{time:YYYY-MM-DD}.log",
rotation="00:00", # 每天午夜轮转
retention="7 days", # 保留7天
encoding="utf-8"
)

class ChromeStoreCrawler:
"""Chrome扩展商店爬虫主控类"""

def __init__(self):
self.fetcher = ChromeStoreFetcher()
self.parser = ChromeStoreParser()
self.cleaner = DataCleaner()
self.storage = DataStorage()

def crawl_category(self, category_url: str, category_name: str = "unknown"):
"""
爬取单个分类页

Args:
category_url: 分类页URL
category_name: 分类名称（用于文件命名）
"""
logger.info(f"🚀 开始爬取分类: {category_name}")
logger.info(f" URL: {category_url}")

# 1. 请求页面
response = self.fetcher.fetch(category_url)
if not response:
logger.error("❌ 页面请求失败")
return

# 2. 解析数据
plugins = self.parser.parse_category_page(response.text)
if not plugins:
logger.error("❌ 未解析到数据")
return

# 3. 清洗数据
cleaned_plugins = self.cleaner.clean_plugins(plugins)

# 4. 补充分类字段
for plugin in cleaned_plugins:
if not plugin.get('category'):
plugin['category'] = category_name

# 5. 保存数据
filename = f"{category_name}_{len(cleaned_plugins)}"
self.storage.save_csv(cleaned_plugins, filename)
self.storage.save_json(cleaned_plugins, filename)

logger.info(f"✅ 爬取完成: {len(cleaned_plugins)} 个插件")

def crawl_multiple_categories(self, categories: dict):
"""
爬取多个分类

Args:
categories: {分类名: URL} 字典
"""
total = len(categories)
for idx, (name, url) in enumerate(categories.items(), 1):
logger.info(f"\\n{'='*60}")
logger.info(f"进度: {idx}/{total}")
logger.info(f"{'='*60}")

self.crawl_category(url, name)

# 间隔时间（除了最后一个）
if idx < total:
logger.info("⏸ 等待5秒…")
import time
time.sleep(5)

logger.info(f"\\n🎉 全部完成！共爬取 {total} 个分类")

def main():
"""主函数"""
crawler = ChromeStoreCrawler()

# 定义要爬取的分类（示例）
categories = {
"productivity": "https://chromewebstore.google.com/category/extensions/productivity",
"developer_tools": "https://chromewebstore.google.com/category/extensions/dev-tools",
"shopping": "https://chromewebstore.google.com/category/extensions/shopping"
}

# 开始爬取
crawler.crawl_multiple_categories(categories)

if __name__ == "__main__":
main()

启动命令

# 方式一：直接运行
python main.py

# 方式二：指定Python版本
python3.10 main.py

# 方式三：后台运行（Linux/Mac）
nohup python main.py > output.log 2>&1 &

# 方式四：使用虚拟环境
source venv/bin/activate # 激活虚拟环境
python main.py

输出结构

data/
├── csv/
│ ├── productivity_25.csv # 生产工具分类，25个插件
│ ├── developer_tools_18.csv # 开发工具分类，18个插件
│ └── shopping_30.csv # 购物分类，30个插件
└── json/
├── productivity_25.json
├── developer_tools_18.json
└── shopping_30.json

logs/
└── crawler_2026–01–27.log # 当天的日志

示例输出（CSV格式）

示例输出（JSON格式）

{
"metadata": {
"total_count": 25,
"export_time": "2026-01-27T14:30:00",
"version": "1.0"
},
"data": [
{
"plugin_id": "nmmhkkegccagdldgiimedpiccmgmieda",
"plugin_name": "AdBlock — 最佳广告拦截工具",
"category": "productivity",
"rating": 4.6,
"rating_count": 125847,
"users_count": 10000000,
"users_count_text": "1000万+ 位用户",
"developer": "AdBlock Inc.",
"detail_url": "https://chromewebstore.google.com/detail/nmmhkkegccagdldgiimedpiccmgmieda",
"description": "阻止YouTube、Facebook等网站上的烦人广告…"
}
]
}

控制台输出（日志）

15:30:15 | INFO | 🚀 开始爬取分类: productivity
15:30:15 | INFO | URL: https://chromewebstore.google.com/category/extensions/productivity
15:30:16 | INFO | ✓ 请求成功: https://chromewebstore.google.com/category/ext… [状态码: 200]
15:30:16 | INFO | 找到 25 个插件卡片
15:30:16 | INFO | ✓ 成功解析 25 个插件
15:30:16 | INFO | ✓ 清洗完成: 24/25 个插件有效
15:30:16 | INFO | ✓ CSV已保存: data/csv/productivity_24.csv (24 行)
15:30:16 | INFO | ==================================================
15:30:16 | INFO | 📊 数据统计报告
15:30:16 | INFO | ==================================================
15:30:16 | INFO | • 总插件数: 24
15:30:16 | INFO | • 分类数: 1
15:30:16 | INFO | • 平均评分: 4.51 星
15:30:16 | INFO | • 高分插件(≥4.0): 22 个 (91.7%)
15:30:16 | INFO | • 总用户数: 45,230,000
15:30:16 | INFO |
15:30:16 | INFO | 🏆 用户数Top 5:
15:30:16 | INFO | AdBlock — 最佳广告拦截工具: 10,000,000 用户
15:30:16 | INFO | Grammarly: Grammar Checker: 5,000,000 用户
15:30:16 | INFO | Honey: Automatic Coupons: 3,500,000 用户
15:30:16 | INFO | ==================================================
15:30:16 | INFO | ✓ JSON已保存: data/json/productivity_24.json (24 条)
15:30:16 | INFO | ✅ 爬取完成: 24 个插件

1️⃣1️⃣ 常见问题与排错（强烈建议写）

Q1: 403 Forbidden – 被识别为爬虫

症状：

ERROR | ✗ 被拒绝访问 (403)：可能被识别为爬虫

原因分析：

解决方案：

# 方案1：使用真实浏览器的完整headers
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Referer': 'https://chromewebstore.google.com/',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
}

# 方案2：增加延迟
time.sleep(random.uniform(3, 8)) # 3-8秒随机延迟

# 方案3：使用代理IP（终极方案）
proxies = {
'http': 'http://your_proxy_ip:port',
'https': 'https://your_proxy_ip:port',
}
response = requests.get(url, proxies=proxies)

代理IP服务推荐：

• 免费：ProxyMesh、HideMyAss（质量不稳定）
• 付费：Bright Data、Oxylabs（稳定但贵）
• 自建：购买云服务器搭建代理池

Q2: 429 Too Many Requests – 触发频率限制

症状：

WARNING | ⚠ 触发限流 (429)，即将重试…

原因：短时间内发送了过多请求

解决方案：

# 增加基础延迟
min_interval = 5 # 从2秒增加到5秒

# 指数退避
if response.status_code == 429:
retry_after = response.headers.get('Retry-After', 60)
logger.warning(f"服务器要求等待 {retry_after} 秒")
time.sleep(int(retry_after))

Q3: 页面结构变化导致解析失败

症状：

WARNING | ⚠ 未找到插件卡片，页面结构可能已变化

原因：

• 网站改版，HTML结构变化
• 选择器失效

排查步骤：

fetcher.save_html(response.text, "debug_page")

# 打开 data/raw/debug_page.html
# 手动查找插件卡片的新class名

# 旧选择器（失效）
plugin_cards = soup.select('div[role="listitem"]')

# 新选择器（根据实际HTML调整）
plugin_cards = soup.select('div.new-plugin-card') # 假设class改成了new-plugin-card

防御性编程：

# 使用多个选择器作为备选方案
selectors = [
'div[role="listitem"]', # 主选择器
'div.plugin-card', # 备选1
'article.extension', # 备选2
]

for selector in selectors:
plugin_cards = soup.select(selector)
if plugin_cards:
logger.info(f"✓ 使用选择器: {selector}")
break
else:
logger.error("❌ 所有选择器均失败")

Q4: 编码乱码问题

症状：CSV文件在Excel中打开显示乱码

原因：编码不匹配

解决方案：

# ✅ 正确做法：使用utf-8-sig编码
df.to_csv('output.csv', encoding='utf-8-sig', index=False)

# ❌ 错误做法：使用utf-8（Excel不识别）
df.to_csv('output.csv', encoding='utf-8', index=False)

为什么utf-8-sig有效？

• utf-8-sig 会在文件开头添加BOM（Byte Order Mark）
• BOM告诉Excel：“这是UTF-8编码的文件”
• Excel看到BOM后会正确解码中文

Q5: 内存溢出（爬取大量数据时）

症状：

MemoryError: Unable to allocate…

原因：一次性加载太多数据到内存

解决方案：分批处理

def crawl_with_pagination(base_url, max_pages=10):
"""分页爬取"""
all_plugins = []

for page in range(1, max_pages + 1):
url = f"{base_url}?page={page}"
plugins = crawl_single_page(url)

# 每爬10页保存一次
all_plugins.extend(plugins)
if page % 10 == 0:
storage.save_csv(all_plugins, f"batch_{page}")
all_plugins = [] # 释放内存

time.sleep(3)

# 保存剩余数据
if all_plugins:
storage.save_csv(all_plugins, f"batch_final")

Q6: 动态内容未加载

症状：爬到的HTML只有loading…，没有实际内容

判断方法：

if '<div>Loading…</div>' in response.text:
logger.warning("⚠ 检测到动态加载页面")

解决方案A：找API接口（推荐）

# 示例：找到的API接口
api_url = "https://chromewebstore.google.com/api/v1/items"
params = {
'category': 'productivity',
'page': 1,
'count': 30
}
response = requests.get(api_url, params=params, headers=headers)
data = response.json()

解决方案B：使用Selenium（备选）

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

# 初始化浏览器
options = webdriver.ChromeOptions()
options.add_argument('–headless') # 无头模式
driver = webdriver.Chrome(options=options)

# 访问页面
driver.get(url)

# 等待内容加载
wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.CLASS_NAME, 'plugin-card')))

# 获取渲染后的HTML
html = driver.page_source

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

并发爬取：提升10倍效率

串行爬取：1个分类 → 2秒 → 10个分类 = 20秒

并发爬取：10个分类同时进行 = 3秒

方案1：使用 concurrent.futures

from concurrent.futures import ThreadPoolExecutor, as_completed

def crawl_with_threads(categories: dict, max_workers=5):
"""
多线程并发爬取

Args:
categories: {分类名: URL} 字典
max_workers: 最大线程数（建议3-5，避免被封）
"""
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# 提交所有任务
future_to_category = {
executor.submit(crawler.crawl_category, url, name): name
for name, url in categories.items()
}

# 处理完成的任务
for future in as_completed(future_to_category):
category = future_to_category[future]
try:
future.result()
logger.info(f"✓ {category} 完成")
except Exception as e:
logger.error(f"✗ {category} 失败: {e}")

优缺点：

• ✅ 优点：代码简单，适合I/O密集型任务
• ⚠️ 注意：线程数不宜过多（会被识别为攻击）

方案2：使用 asyncio + aiohttp

import asyncio
import aiohttp

class AsyncCrawler:
async def fetch_async(self, session, url):
"""异步请求"""
async with session.get(url, headers=self.headers) as response:
return await response.text()

async def crawl_multiple_async(self, urls: list):
"""异步爬取多个URL"""
async with aiohttp.ClientSession() as session:
tasks = [self.fetch_async(session, url) for url in urls]
htmls = await asyncio.gather(*tasks)
return htmls

# 使用
crawler = AsyncCrawler()
urls = ["url1", "url2", "url3"]
htmls = asyncio.run(crawler.crawl_multiple_async(urls))

优缺点：

• ✅ 优点：性能最高，单线程处理大量请求
• ❌ 缺点：代码复杂，学习曲线陡峭

断点续跑：避免重复劳动

场景：爬了500个插件后程序崩溃，重新运行又从头开始？

解决方案：检查点机制

import pickle

class Checkpoint:
"""检查点管理器"""

def __init__(self, filename='checkpoint.pkl'):
self.filename = filename
self.crawled_ids = self.load()

def load(self):
"""加载已爬取的ID"""
try:
with open(self.filename, 'rb') as f:
return pickle.load(f)
except FileNotFoundError:
return set()

def save(self, plugin_id):
"""保存一个ID"""
self.crawled_ids.add(plugin_id)
with open(self.filename, 'wb') as f:
pickle.dump(self.crawled_ids, f)

def is_crawled(self, plugin_id):
"""检查是否已爬取"""
return plugin_id in self.crawled_ids

# 在主程序中使用
checkpoint = Checkpoint()

for category_name, url in categories.items():
if checkpoint.is_crawled(category_name):
logger.info(f"⏭ 跳过已爬取的分类: {category_name}")
continue

# 爬取
crawler.crawl_category(url, category_name)

# 标记为已完成
checkpoint.save(category_name)

增量更新：只爬新数据

场景：每周更新一次数据，但不想重复爬取旧插件

解决方案：

def incremental_crawl(storage, category_url):
"""增量爬取"""
# 1. 加载已有数据
existing_df = storage.load_csv('productivity_old')
existing_ids = set(existing_df['plugin_id'].tolist())

logger.info(f"已有 {len(existing_ids)} 个插件")

# 2. 爬取最新数据
new_plugins = crawler.crawl_category(category_url)

# 3. 过滤：只保留新插件
new_plugins_filtered = [
p for p in new_plugins
if p['plugin_id'] not in existing_ids
]

logger.info(f"发现 {len(new_plugins_filtered)} 个新插件")

# 4. 合并并保存
all_plugins = existing_df.to_dict('records') + new_plugins_filtered
storage.save_csv(all_plugins, 'productivity_updated')

监控与告警：自动检测异常

class CrawlerMonitor:
"""爬虫监控器"""

def __init__(self):
self.success_count = 0
self.fail_count = 0
self.start_time = time.time()

def record_success(self):
self.success_count += 1

def record_failure(self):
self.fail_count += 1

def get_success_rate(self):
total = self.success_count + self.fail_count
return (self.success_count / total * 100) if total > 0 else 0

def alert_if_needed(self):
"""成功率低于50%时发送告警"""
success_rate = self.get_success_rate()

if success_rate < 50:
logger.error(f"🚨 告警：成功率仅 {success_rate:.1f}%，可能被封禁！")
# 可以在这里发送邮件/短信/钉钉通知
self.send_alert(f"爬虫成功率低: {success_rate:.1f}%")

def send_alert(self, message):
"""发送告警（示例：打印到日志）"""
logger.critical(f"📧 告警消息: {message}")
# 实际应用中可以接入：
# – 邮件：smtplib
# – 钉钉：requests.post(webhook_url)
# – 企业微信、Slack等

定时任务：自动化运行

方案1：Linux Cron

# 编辑crontab
crontab -e

# 每天凌晨2点运行
0 2 * * * cd /path/to/project && /usr/bin/python3 main.py >> logs/cron.log 2>&1

# 每周一上午10点运行
0 10 * * 1 cd /path/to/project && /usr/bin/python3 main.py

方案2：Python APScheduler

from apscheduler.schedulers.blocking import BlockingScheduler

def scheduled_crawl():
"""定时任务函数"""
logger.info("⏰ 定时任务启动")
crawler = ChromeStoreCrawler()
crawler.crawl_multiple_categories(categories)
logger.info("✅ 定时任务完成")

# 创建调度器
scheduler = BlockingScheduler()

# 添加任务：每天凌晨2点执行
scheduler.add_job(scheduled_crawl, 'cron', hour=2, minute=0)

# 添加任务：每小时执行一次
scheduler.add_job(scheduled_crawl, 'interval', hours=1)

# 启动调度器
logger.info("⏲ 调度器已启动")
scheduler.start()

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

我们完成了什么？

在这篇教程中，我们从零开始构建了一个生产级别的Chrome扩展商店爬虫，涵盖了爬虫开发的完整流程：

✅ 技术栈选择 选择了 requests + BeautifulSoup + pandas 的轻量级组合，避免了Selenium的性能开销，同时保证了稳定性和可维护性。

✅ 工程化实践 采用了模块化设计（Fetcher/Parser/Cleaner/Storage四层架构），每个模块职责单一，便于测试和扩展。

✅ 反爬对抗 实现了User-Agent随机化、频率控制、指数退避重试、Session复用等多重反爬策略，显著降低了被封禁风险。

✅ 数据质量保障 通过多层数据清洗（原始文本 → 初步清洗 → 格式转换 → 去重验证），确保了输出数据的准确性和一致性。

✅ 可扩展性 支持并发爬取、断点续跑、增量更新、监控告警等高级功能，可以轻松应对大规模数据采集需求。

这个项目的真实价值

坦白说，写这篇教程最大的收获不是代码本身，而是解决问题的思维方式：

很多初学者写爬虫时的常见误区：

• ❌ 遇到403就放弃 → ✅ 分析原因，逐一排查headers、频率、代理
• ❌ 页面结构变化就重写 → ✅ 设计多套选择器备选方案，增加鲁棒性
• ❌ 数据直接存数据库 → ✅ 先存CSV/JSON，便于调试和回滚

还有哪些可以优化的地方？

1. 分布式爬取

• 使用 Scrapy-Redis 实现多机协同
• 搭建任务队列（RabbitMQ/Celery）分发爬取任务
• 适用场景：需要爬取百万级数据

2. 智能代理池

• 自动检测代理IP可用性
• 动态切换失效代理
• 集成第三方代理服务API

3. 验证码识别

• 使用 OCR 库（Tesseract）识别简单验证码
• 接入打码平台（2Captcha）处理复杂验证码
• 使用深度学习模型（YOLO）识别图形验证码

4. 数据可视化

• 使用 Plotly/Echarts 生成交互式图表
• 统计各分类的插件数量分布
• 分析评分与下载量的相关性

5. 接入数据库

# 示例：存储到MySQL
import pymysql

conn = pymysql.connect(host='localhost', user='root', password='xxx', database='plugins')
cursor = conn.cursor()

for plugin in plugins:
sql = "INSERT INTO plugins (id, name, rating) VALUES (%s, %s, %s)"
cursor.execute(sql, (plugin['plugin_id'], plugin['plugin_name'], plugin['rating']))

conn.commit()

延伸阅读推荐

📚 书籍：

• 《Python网络数据采集》（Web Scraping with Python）by Ryan Mitchell
• 《Python爬虫开发与项目实战》by 范传辉

… …

🌟 文末

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

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

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

专栏 👉 《Python爬虫实战》，我会按照“入门 → 进阶 → 工程化 → 项目落地”的路线持续更新，争取让每一篇都做到：

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

📣 想系统提升的小伙伴：强烈建议先订阅专栏，再按目录顺序学习，效率会高很多～

✅ 互动征集

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

评论区留言告诉我你的需求，我会优先安排更新 ✅

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

免责声明：本文仅用于学习与技术研究，请在合法合规、遵守站点规则与 Robots 协议的前提下使用相关技术。严禁将技术用于任何非法用途或侵害他人权益的行为。技术无罪，责任在人！！！