Python 异步爬虫实战：FindQC 商品数据爬取系统完整教程

本文详细介绍如何使用 Python 异步编程技术构建一个高性能的商品数据爬虫系统，包括 API 调用、数据库存储、消息队列集成等核心功能。

📋 目录

• 一、项目概述

• 二、技术栈

• 三、项目架构

• 四、核心功能

• 五、环境配置

• 六、代码详解

• 七、使用示例

• 八、性能优化

• 九、常见问题

• 十、总结

一、项目概述

1.1 项目简介

service_spider 是一个基于 Python 异步编程的商品数据爬虫服务，主要功能包括：

• ✅ 目录遍历：自动遍历所有需要爬取的目录

• ✅ 分页处理：智能分页获取商品列表，直到最后一页

• ✅ 商品详情获取：获取商品基本信息、图集（QC图、视频等）

• ✅ 数据存储：保存商品数据到 MySQL 数据库

• ✅ 消息队列：发送商品新增消息到 RabbitMQ，通知下游服务

• ✅ 断点续传：支持中断后从断点继续爬取

• ✅ 优雅中断：支持 Ctrl+C 优雅退出，确保数据安全

1.2 项目结构

service_spider/
├── __init__.py # 包初始化
├── main.py # 主程序入口（一次性执行）
├── scheduler.py # 定时任务入口（使用 APScheduler）
├── spider.py # 爬虫核心逻辑
├── api_client.py # FindQC API 客户端封装
├── db_service.py # 数据库操作服务
├── crawl_from_json.py # 从 JSON 文件爬取（辅助功能）
├── README.md # 项目文档
├── TEST.md # 测试指南
└── CSDN博客教程.md # 本文档

二、技术栈

2.1 核心技术

2.2 设计模式

• 生产者-消费者模式：目录并发爬取，商品并发处理

• 严格漏桶限流：极致平滑的 API 调用控制

• 优雅中断机制：支持 Ctrl+C 安全退出

• 断点续传：自动从上次中断位置继续

三、项目架构

3.1 架构图

┌─────────────────────────────────────────────────────────┐
│ service_spider │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ main.py │───▶│ spider.py │ │
│ │ (入口) │ │ (核心逻辑) │ │
│ └──────────────┘ └──────────────┘ │
│ │ │ │
│ │ ▼ │
│ │ ┌──────────────┐ │
│ │ │ api_client.py │ │
│ │ │ (API调用) │ │
│ │ └──────────────┘ │
│ │ │ │
│ │ ▼ │
│ │ ┌──────────────┐ │
│ └───────────▶│ db_service.py│ │
│ │ (数据库) │ │
│ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
│ │ │
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ FindQC │ │ MySQL │ │RabbitMQ │
│ API │ │Database │ │ Queue │
└─────────┘ └─────────┘ └─────────┘

3.2 数据流

FindQC API
↓
商品列表 (getCategoryProducts)
↓
商品详情 (get_product_detail)
↓
商品图集 (get_product_atlas)
↓
数据整理 (prepare_product_data)
↓
MySQL (t_products, t_tasks_products)
↓
RabbitMQ (product.new 消息)

3.3 核心流程

开始
↓
获取目录列表
↓
for 每个目录（并发）:
↓
分页获取商品列表 (while True)
↓
for 每个商品（并发）:
↓
获取商品详情
↓
获取商品图集（分页）
↓
整理图片结构（QC图、主图、SKU图）
↓
保存到数据库
↓
创建任务记录
↓
发送消息到 RabbitMQ
↓
end
↓
判断是否最后一页
↓
否 → 翻页继续
↓
是 → 下一个目录
↓
end
↓
完成

四、核心功能

4.1 目录遍历

爬虫会自动遍历配置的目录列表，支持并发处理多个目录，提高爬取效率。

关键代码：

async def get_target_categories(self) -> List[Dict[str, Any]]:
"""获取目标目录列表"""
# 从配置或数据库读取目录列表
categories = [
{"catalogueId": 4113, "name": "测试分类"},
# … 更多目录
]
return categories

4.2 分页处理

智能分页获取商品列表，直到 hasMore=False 表示最后一页。

关键代码：

page = 1
while True:
response = await self.api_client.get_category_products(
catalogue_id=category_id,
page=page,
size=self.page_size
)

if not response.get("hasMore", False):
break # 最后一页

products = response.get("data", [])
# 处理商品…
page += 1

4.3 商品详情获取

获取商品的完整信息，包括：

• 基本信息（名称、价格、描述等）

• 主图列表

• SKU 信息

• QC 图集（分页获取）

关键代码：

# 获取商品详情
detail = await self.api_client.get_product_detail(
item_id=item_id,
mall_type=mall_type
)

# 获取商品图集（分页）
atlas_page = 1
while True:
atlas_response = await self.api_client.get_product_atlas(
goods_id=goods_id,
item_id=item_id,
mall_type=mall_type,
page=atlas_page,
size=50
)

if not atlas_response.get("hasMore", False):
break

atlas_page += 1

4.4 数据存储

将商品数据保存到 MySQL 数据库，包括：

• t_products 表：商品基本信息

• t_tasks_products 表：任务记录

关键代码：

# 检查商品是否已存在
existing_product, operation = await ProductDBService.check_and_update_existing_product(
session=session,
findqc_id=findqc_id,
update_task_id=update_task_id,
last_qc_time=last_qc_time,
qc_count_3days=qc_count_3days,
# … 其他参数
)

if operation == "not_exists":
# 新商品：保存完整数据
product = await ProductDBService.save_or_update_product(
session=session,
product_data=product_data,
update_task_id=update_task_id
)

# 创建任务记录
task = await ProductDBService.create_task(
session=session,
product_id=product.id,
update_task_id=update_task_id
)

4.5 消息队列

发送商品新增消息到 RabbitMQ，通知下游 AI 处理管道。

消息格式：

{
"task_id": 2024052001,
"findqc_id": 12345,
"product_id": 1001,
"itemId": "ext_999",
"mallType": "taobao",
"action": "product.new",
"timestamp": "2024-05-20T10:00:00Z"
}

4.6 断点续传

支持中断后从断点继续爬取，避免重复工作。

实现原理：

关键代码：

# 检查断点续传
resume_category_id = await ProductDBService.get_resume_category_id(
session=session,
today_task_id=update_task_id,
)

if resume_category_id is not None:
start_cat_id = resume_category_id
logger.info(f"启用断点续传：将从目录ID {start_cat_id} 重新开始爬取")

4.7 优雅中断

支持 Ctrl+C 优雅退出，确保数据安全和资源清理。

实现原理：

关键代码：

class GracefulShutdown:
"""优雅中断管理器"""

def __init__(self):
self.shutdown_requested = False
self.shutdown_event = asyncio.Event()

def request_shutdown(self):
"""请求关闭"""
self.shutdown_requested = True
self.shutdown_event.set()
logger.warning("🛑 收到中断信号，正在优雅退出…")

def setup_signal_handlers(self):
"""设置信号处理器"""
def signal_handler(signum, frame):
if self.shutdown_requested:
sys.exit(1) # 强制退出
else:
self.request_shutdown()

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

五、环境配置

5.1 安装依赖

# 创建虚拟环境
python3 -m venv venv
source venv/bin/activate # Linux/Mac
# 或
venv\\Scripts\\activate # Windows

# 安装依赖
pip install sqlalchemy aiomysql loguru httpx pydantic-settings
pip install playwright apscheduler pika

# 安装 Playwright 浏览器
playwright install chromium

5.2 环境变量配置

创建 .env 文件：

# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=your_password
DB_NAME=findqc_db

# FindQC API 配置
FINDQC_API_BASE_URL=https://api.findqc.com
FINDQC_API_KEY=your_api_key # 可选

# RabbitMQ 配置
RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/

# 日志级别
LOG_LEVEL=INFO

# 测试模式：只爬取 N 个商品（0 或不设置表示全量模式）
MAX_PRODUCTS=0

# 并发配置
MAX_CONCURRENT_CATEGORIES=5 # 最大并发目录数
MAX_CONCURRENT_PRODUCTS_PER_CATALOGUE=10 # 每个目录内商品并发数

# QPS 限流配置
CATEGORY_PRODUCTS_MAX_QPS=10 # 目录商品列表 API QPS
PRODUCT_DETAIL_MAX_QPS=15 # 商品详情 API QPS
PRODUCT_ATLAS_MAX_QPS=20 # 商品图集 API QPS

5.3 数据库初始化

— 创建数据库
CREATE DATABASE findqc_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

— 表结构会自动创建（通过 db.init_db()）

六、代码详解

6.1 API 客户端（api_client.py）

6.1.1 严格漏桶限流器

核心原理：强制每个请求之间间隔固定时间，实现极致平滑的 API 调用。

class StrictRateLimiter:
"""严格的漏桶限流器（Leaky Bucket）"""

def __init__(self, max_qps: float, window_size: int = 30):
self.max_qps = max_qps
self.interval = 1.0 / max_qps if max_qps > 0 else 0.0 # 请求间隔
self.next_allow_time = 0.0 # 下一次允许请求的时间
self.lock = asyncio.Lock()

async def acquire(self):
"""获取执行许可，包含平滑控制"""
async with self.lock:
now = time.time()
if self.next_allow_time < now:
self.next_allow_time = now

# 我的执行时间就是 next_allow_time
my_schedule_time = self.next_allow_time

# 将下一次允许时间向后推一个间隔
self.next_allow_time += self.interval

# 计算需要等待的时间
wait_time = my_schedule_time – time.time()
if wait_time > 0:
await asyncio.sleep(wait_time)

使用示例：

# 初始化限流器（QPS=10，即每秒最多10个请求）
limiter = StrictRateLimiter(max_qps=10)

# 在 API 调用前获取许可
await limiter.acquire()
response = await session.get(url)

6.1.2 Playwright 浏览器自动化

核心功能：使用 Playwright 绕过 Cloudflare 反爬虫保护。

class FindQCAPIClient:
"""FindQC API 客户端"""

async def _init_browser(self):
"""初始化浏览器"""
if not HAS_PLAYWRIGHT:
raise ImportError("Playwright 未安装")

self.playwright = await async_playwright().start()
self.browser = await self.playwright.chromium.launch(
headless=True,
args=['–disable-blink-features=AutomationControlled']
)
self.context = await self.browser.new_context(
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
viewport={'width': 1920, 'height': 1080}
)
self.page = await self.context.new_page()

async def get_category_products(self, catalogue_id: int, page: int, size: int = 20):
"""获取目录商品列表"""
# 使用限流器控制请求频率
await self.category_products_strict_limiter.acquire()

# 使用 Playwright 访问页面，绕过 Cloudflare
url = f"{self.base_url}/goods/getCategoryProducts"
await self.page.goto(url)

# 监听网络请求，捕获 API 响应
response_data = await self._wait_for_api_response(url_pattern)

return response_data

6.2 爬虫核心逻辑（spider.py）

6.2.1 商品处理逻辑

核心流程：

async def process_product(self, product_info: Dict[str, Any], category_id: int):
"""处理单个商品"""
try:
# 1. 获取商品详情
detail = await self.api_client.get_product_detail(…)

# 2. 获取商品图集（分页）
all_qc_images = []
atlas_page = 1
while True:
atlas_response = await self.api_client.get_product_atlas(…)
all_qc_images.extend(atlas_response.get("data", []))
if not atlas_response.get("hasMore", False):
break
atlas_page += 1

# 3. 整理商品数据
product_data = self.prepare_product_data(detail, all_qc_images)

# 4. 保存到数据库
async with self.db.async_session_maker() as session:
# 检查商品是否已存在
existing_product, operation = await ProductDBService.check_and_update_existing_product(…)

if operation == "not_exists":
# 新商品：保存完整数据
product = await ProductDBService.save_or_update_product(…)
task = await ProductDBService.create_task(…)

# 发送消息到 RabbitMQ
await self.send_product_message(…)
elif operation == "exists_updated":
# 已存在：只更新 QC 相关字段
await session.commit()
elif operation == "exists_deleted":
# 已存在：软删除
await session.commit()

except Exception as e:
logger.error(f"处理商品失败: {e}")

6.2.2 并发控制

目录并发：

async def spider_main_process(self, update_task_id: int, max_products: Optional[int] = None):
"""爬虫主流程"""
categories = await self.get_target_categories()

# 使用信号量控制并发目录数
semaphore = asyncio.Semaphore(self.max_concurrent_categories)

async def process_category(category):
async with semaphore:
await self.process_category(category, update_task_id, max_products)

# 并发处理所有目录
tasks = [process_category(cat) for cat in categories]
await asyncio.gather(*tasks, return_exceptions=True)

商品并发：

async def process_category(self, category: Dict[str, Any], update_task_id: int, max_products: Optional[int] = None):
"""处理单个目录"""
page = 1
products_processed = 0

while True:
# 获取商品列表
response = await self.api_client.get_category_products(…)
products = response.get("data", [])

# 使用信号量控制并发商品数
semaphore = asyncio.Semaphore(self.max_concurrent_products_per_catalogue)

async def process_product_safe(product_info):
async with semaphore:
await self.process_product(product_info, category_id)

# 并发处理商品
tasks = [process_product_safe(p) for p in products]
await asyncio.gather(*tasks, return_exceptions=True)

if not response.get("hasMore", False):
break

page += 1

6.3 数据库服务（db_service.py）

6.3.1 商品存在性检查

核心逻辑：根据最新爬取的 QC 图时间判断商品状态。

@staticmethod
async def check_and_update_existing_product(
session: AsyncSession,
findqc_id: int,
update_task_id: int,
last_qc_time: Optional[datetime],
qc_count_360days: int,
# … 其他参数
) -> Tuple[Optional[Product], str]:
"""检查现有商品并决定更新策略"""

# 查询商品是否已存在
product = await session.execute(
select(Product).where(Product.findqc_id == findqc_id)
).scalar_one_or_none()

if not product:
return None, "not_exists"

# 判断最新爬取的QC图是否在360天内
thirty_days_ago = datetime.utcnow() – timedelta(days=360)

if last_qc_time is None or last_qc_time < thirty_days_ago:
# QC图不在360天内，软删除
product.status = 1
product.qc_count_3days = 0
product.qc_count_7days = 0
# … 其他字段设为0
return product, "exists_deleted"
else:
# QC图在360天内，更新QC相关字段并重新拾取
product.last_qc_time = last_qc_time
product.qc_count_3days = qc_count_3days
product.qc_count_7days = qc_count_7days
# … 更新其他QC字段
product.status = 0 # 重新拾取
return product, "exists_updated"

6.3.2 新商品保存

@staticmethod
async def save_or_update_product(
session: AsyncSession,
product_data: Dict[str, Any],
update_task_id: int,
) -> Product:
"""保存新商品数据"""

# 创建新商品对象
product = Product(
findqc_id=product_data["findqc_id"],
item_id=product_data["item_id"],
mall_type=product_data["mall_type"],
name=product_data["name"],
price=product_data["price"],
main_img=product_data["main_img"],
# … 其他字段
update_task_id=update_task_id,
status=0,
)

session.add(product)
await session.flush() # 获取 product.id

return product

七、使用示例

7.1 一次性执行

# 激活虚拟环境
source venv/bin/activate

# 运行爬虫服务（执行一次后退出）
python -m service_spider.main

7.2 定时任务模式

# 激活虚拟环境
source venv/bin/activate

# 运行定时任务服务（持续运行，按配置的时间执行爬虫任务）
python -m service_spider.scheduler

定时任务配置（通过环境变量）：

# Cron 模式：每天 02:00 执行
SPIDER_SCHEDULE_TYPE=cron
SPIDER_CRON_HOUR=2
SPIDER_CRON_MINUTE=0

# Interval 模式：每 24 小时执行一次
# SPIDER_SCHEDULE_TYPE=interval
# SPIDER_INTERVAL_HOURS=24

7.3 测试模式

# 设置环境变量：只爬取 10 个商品
export MAX_PRODUCTS=10

# 运行爬虫
python -m service_spider.main

7.4 查看日志

# 控制台输出（实时）
# 日志会自动输出到控制台

# 文件日志（按天轮转）
tail -f logs/spider_2024-05-20.log

八、性能优化

8.1 并发控制

• 目录并发：使用 asyncio.Semaphore 控制并发目录数（默认 5）

• 商品并发：使用 asyncio.Semaphore 控制每个目录内商品并发数（默认 10）

配置建议：

# 根据 API 限流和服务器性能调整
MAX_CONCURRENT_CATEGORIES=5 # 目录并发数
MAX_CONCURRENT_PRODUCTS_PER_CATALOGUE=10 # 商品并发数

8.2 QPS 限流

使用严格漏桶限流器实现极致平滑的 API 调用：

# 不同 API 使用不同的 QPS 限制
CATEGORY_PRODUCTS_MAX_QPS=10 # 目录商品列表 API
PRODUCT_DETAIL_MAX_QPS=15 # 商品详情 API
PRODUCT_ATLAS_MAX_QPS=20 # 商品图集 API

8.3 数据库优化

• 批量提交：每个商品独立事务，失败不影响其他商品

• 索引优化：在 findqc_id 字段上创建唯一索引

• 连接池：使用 SQLAlchemy 连接池管理数据库连接

8.4 内存管理

• 分批处理：使用分页获取商品列表，避免一次性加载过多数据

• 及时释放：处理完的商品数据及时释放，避免内存泄漏

九、常见问题

9.1 ModuleNotFoundError

问题：ModuleNotFoundError: No module named 'sqlalchemy'

解决：安装依赖

pip install -r requirements.txt

9.2 数据库连接失败

问题：Can't connect to MySQL server

解决：

• 检查 MySQL 服务是否启动

• 检查 .env 文件中的数据库配置是否正确

• 确认数据库用户有创建表的权限

9.3 API 请求失败（429 限流）

问题：HTTPStatusError: 429 Too Many Requests

解决：

• 增加请求延迟时间（修改 delay_between_requests 参数）

• 降低 QPS 限制（修改 CATEGORY_PRODUCTS_MAX_QPS 等配置）

• 减少并发数（修改 MAX_CONCURRENT_CATEGORIES 等配置）

9.4 Playwright 浏览器启动失败

问题：Playwright 未安装

解决：

# 安装 Playwright
pip install playwright

# 安装浏览器
playwright install chromium

9.5 断点续传不生效

问题：断点续传没有从上次中断位置继续

解决：

• 检查数据库中是否有今天（update_task_id = 当天日期）的数据

• 确认 get_resume_category_id 方法能正确查询到最小 catalogueId

9.6 消息队列发送失败

问题：RabbitMQ 消息发送失败

解决：

• 检查 RabbitMQ 服务是否启动

• 检查 .env 文件中的 RabbitMQ 配置是否正确

• 如果未配置 RabbitMQ，爬虫仍然可以运行（只是不会发送消息）

十、总结

10.1 项目亮点

10.2 适用场景

• 商品数据爬取

• 电商平台数据采集

• API 数据同步

• 定时数据更新

10.3 后续优化方向

• 支持更多数据源

• 添加监控和统计功能

• 优化图片数据处理性能

• 支持分布式爬取

📚 参考资料

• Python asyncio 官方文档

• Playwright 官方文档

• SQLAlchemy 官方文档

• APScheduler 官方文档

作者：MadPrinter 最后更新：2025-12-26 项目地址：GitHub

💡 提示：如果本文对你有帮助，欢迎点赞、收藏、转发！如有问题，欢迎在评论区留言讨论。