使用 Scrapy 框架定制爬虫中间件接入淘宝 API 采集商品数据

一、引言

在电商数据分析、市场调研等领域，获取淘宝平台上的商品数据是一项常见需求。淘宝提供了 API 接口，允许开发者通过授权的方式获取商品信息。本文将介绍如何使用 Scrapy 框架定制爬虫中间件，实现对淘宝 API 的接入，从而高效地采集商品数据。

二、淘宝 API 概述

淘宝提供了丰富的 API 接口，涵盖商品、交易、用户、营销等多个领域。对于商品数据采集，常用的 API 包括：

taobao.search：搜索商品列表
taobao.item.get：获取单个商品详情
taobao.cats.get：获取商品分类
taobao.props.get：获取商品属性

使用淘宝 API 需要先注册账号并获取 ApiKey和 ApiSecret。同时，API 调用有频率限制，需要合理控制请求速度。

三、Scrapy 框架与中间件

import logging
import time
import random
import hashlib
import hmac
import json
from urllib.parse import urlencode
import requests
from scrapy import signals
from scrapy.exceptions import NotConfigured, IgnoreRequestclass TaobaoAPIMiddleware:"""淘宝API爬虫中间件，处理API请求和响应"""def __init__(self, app_key, app_secret, api_url, rate_limit, retry_times):self.app_key = app_keyself.app_secret = app_secretself.api_url = api_urlself.rate_limit = rate_limitself.retry_times = retry_timesself.request_count = 0self.last_reset_time = time.time()self.logger = logging.getLogger(__name__)@classmethoddef from_crawler(cls, crawler):"""从配置中获取中间件设置"""app_key = crawler.settings.get('TAOBAO_APP_KEY')app_secret = crawler.settings.get('TAOBAO_APP_SECRET')api_url = crawler.settings.get('TAOBAO_API_URL', 'https://eco.taobao.com/router/rest')rate_limit = crawler.settings.getint('TAOBAO_RATE_LIMIT', 500)  # 默认500次/小时retry_times = crawler.settings.getint('TAOBAO_RETRY_TIMES', 3)  # 默认重试3次if not app_key or not app_secret:raise NotConfigured("淘宝API配置缺失：TAOBAO_APP_KEY 和 TAOBAO_APP_SECRET")middleware = cls(app_key, app_secret, api_url, rate_limit, retry_times)crawler.signals.connect(middleware.spider_closed, signal=signals.spider_closed)return middlewaredef process_request(self, request, spider):"""处理API请求，生成签名并发送"""# 检查是否为淘宝API请求if not request.meta.get('taobao_api', False):return None# 检查速率限制self._check_rate_limit()# 构建API请求参数method = request.meta.get('taobao_method')if not method:self.logger.error("淘宝API请求缺少方法名：taobao_method")raise IgnoreRequestparams = self._build_common_params()params.update({'method': method,**request.meta.get('taobao_params', {})})# 生成签名params['sign'] = self._generate_sign(params)try:# 发送API请求response = requests.post(self.api_url,data=params,timeout=30)response.raise_for_status()# 解析JSON响应result = response.json()request.meta['api_result'] = result# 检查API返回状态if not self._check_api_success(result):error_code = result.get('error_response', {}).get('code')error_msg = result.get('error_response', {}).get('msg')self.logger.warning(f"淘宝API返回错误: {error_code} - {error_msg}")# 如果是限流错误，暂停一段时间if error_code in ('isp.over-quota', 'isp.access-control'):self.logger.warning("API请求达到限流阈值，暂停60秒")time.sleep(60)# 重试机制retry_times = request.meta.get('taobao_retry_times', 0)if retry_times < self.retry_times:self.logger.info(f"准备重试API请求，当前重试次数: {retry_times + 1}")new_request = request.copy()new_request.dont_filter = Truenew_request.meta['taobao_retry_times'] = retry_times + 1# 随机延迟后重试time.sleep(random.uniform(1, 3))return new_requestelse:self.logger.error("API请求重试次数已达上限")raise IgnoreRequestexcept requests.exceptions.RequestException as e:self.logger.error(f"API请求发生异常: {str(e)}")# 异常情况下的重试逻辑retry_times = request.meta.get('taobao_retry_times', 0)if retry_times < self.retry_times:self.logger.info(f"准备重试API请求，当前重试次数: {retry_times + 1}")new_request = request.copy()new_request.dont_filter = Truenew_request.meta['taobao_retry_times'] = retry_times + 1# 指数退避策略time.sleep(2 ** retry_times + random.random())return new_requestelse:self.logger.error("API请求重试次数已达上限")raise IgnoreRequest# 处理正常响应return Nonedef _build_common_params(self):"""构建API公共参数"""return {'app_key': self.app_key,'format': 'json','v': '2.0','sign_method': 'hmac','timestamp': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}def _generate_sign(self, params):"""生成API请求签名"""# 排序参数sorted_params = sorted(params.items(), key=lambda x: x[0])# 拼接参数string_to_sign = self.app_secretfor k, v in sorted_params:string_to_sign += f"{k}{v}"string_to_sign += self.app_secret# HMAC-SHA1加密sign = hmac.new(self.app_secret.encode('utf-8'),string_to_sign.encode('utf-8'),hashlib.sha1).hexdigest().upper()return signdef _check_api_success(self, result):"""检查API响应是否成功"""if 'error_response' in result:return False# 根据具体API调整检查逻辑method_name = result.get('method')if method_name:# 例如：taobao.items.list.get 返回值检查if method_name == 'taobao.items.list.get' and 'items_list_response' in result:return True# 其他API方法的检查逻辑...return Truedef _check_rate_limit(self):"""检查API请求速率限制"""current_time = time.time()# 如果已经过了1小时，重置计数器if current_time - self.last_reset_time > 3600:self.request_count = 0self.last_reset_time = current_time# 检查是否超过速率限制if self.request_count >= self.rate_limit:wait_time = 3600 - (current_time - self.last_reset_time)self.logger.warning(f"达到API速率限制，等待{wait_time:.2f}秒")time.sleep(wait_time)# 重置计数器self.request_count = 0self.last_reset_time = current_time# 增加请求计数self.request_count += 1def spider_closed(self, spider):"""爬虫关闭时的清理工作"""self.logger.info("淘宝API中间件: 爬虫已关闭")

Scrapy 是一个为了爬取网站数据、提取结构性数据而编写的应用框架。它可以应用在数据挖掘、信息处理或存储历史数据等一系列的程序中。

Scrapy 的架构中，中间件是一个很重要的组件，分为下载中间件 (Downloader Middleware) 和爬虫中间件 (Spider Middleware)。下载中间件可以拦截请求和响应，进行预处理；爬虫中间件则可以处理爬虫的输入 (响应) 和输出 (请求)。

在接入淘宝 API 的场景中，我们可以定制下载中间件，专门处理 API 请求的签名生成、速率控制、错误重试等逻辑，使爬虫代码更加简洁和可维护。