使用 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 請求的簽名生成、速率控制、錯誤重試等邏輯，使爬蟲代碼更加簡潔和可維護。