在当今的电商数据分析领域，高效且合规地获取商品数据至关重要。京东（JD）作为中国领先的电商平台，其商品数据具有极高的商业价值。本文将介绍如何将 Scrapy 框架与 JD API 集成，实现一种高效且合规的商品数据采集方案。
方案背景与优势
传统的网页爬虫存在诸多问题，包括：
容易触发网站反爬机制，导致 IP 被封
网页结构变化频繁，需要频繁维护爬虫
可能存在法律合规风险
数据提取效率低，尤其是对于 JavaScript 动态加载的内容
相比之下，使用API 结合 Scrapy 框架的方案具有以下优势：
合规性高，符合平台的数据使用规范
数据结构稳定，减少维护成本
采集效率高，支持批量获取
数据完整性好，可获取 API 开放的所有字段
准备工作
1. 申请 JD API 访问权限
首先需要注册创建申请相应的 API 权限。具体步骤请参考api文档。
2. 环境搭建
确保已安装必要的依赖：
pip install scrapy requests python-jose
实现方案
1. 项目结构
plaintext
jd_spider/ ├── jd_spider/ │ ├── __init__.py │ ├── items.py │ ├── middlewares.py │ ├── pipelines.py │ ├── settings.py │ └── spiders/ │ ├── __init__.py │ └── jd_product_spider.py └── scrapy.cfg
2. 核心实现
首先定义数据模型：
import scrapy class JdProductItem(scrapy.Item): # 商品基本信息 product_id = scrapy.Field() # 商品ID name = scrapy.Field() # 商品名称 price = scrapy.Field() # 商品价格 original_price = scrapy.Field() # 原价 brand = scrapy.Field() # 品牌 # 库存与销量 stock = scrapy.Field() # 库存 sales_count = scrapy.Field() # 销量 # 分类信息 category = scrapy.Field() # 分类 category_id = scrapy.Field() # 分类ID # 其他信息 shop_name = scrapy.Field() # 店铺名称 shop_id = scrapy.Field() # 店铺ID comments_count = scrapy.Field() # 评论数 score = scrapy.Field() # 评分 url = scrapy.Field() # 商品链接 crawl_time = scrapy.Field() # 爬取时间
接下来实现 API 请求中间件，处理认证和请求头：
import time
import hashlib
from scrapy import signals
from scrapy.downloadermiddlewares.useragent import UserAgentMiddleware
import random
class JdApiAuthMiddleware:
"""
京东API认证中间件，负责处理API请求的签名和认证
"""
def __init__(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
@classmethod
def from_crawler(cls, crawler):
return cls(
app_key=crawler.settings.get('JD_APP_KEY'),
app_secret=crawler.settings.get('JD_APP_SECRET')
)
def process_request(self, request, spider):
# 只处理JD API的请求
if 'api.jd.com' in request.url:
# 生成时间戳
timestamp = str(int(time.time() * 1000))
# 获取请求参数
params = request.meta.get('params', {})
# 添加必要参数
params['app_key'] = self.app_key
params['timestamp'] = timestamp
params['format'] = 'json'
params['v'] = '1.0'
params['sign_method'] = 'md5'
# 生成签名
sign = self.generate_sign(params)
params['sign'] = sign
# 更新请求URL
request._set_url(self.build_url(request.url, params))
def generate_sign(self, params):
"""生成签名"""
# 按照参数名ASCII升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 拼接参数
sign_str = self.app_secret
for k, v in sorted_params:
sign_str += f"{k}{v}"
sign_str += self.app_secret
# MD5加密并转为大写
return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
def build_url(self, base_url, params):
"""构建完整的URL"""
params_str = '&'.join([f"{k}={v}" for k, v in params.items()])
return f"{base_url}?{params_str}"
class RandomUserAgentMiddleware(UserAgentMiddleware):
"""随机User-Agent中间件"""
def __init__(self, user_agent_list):
self.user_agent_list = user_agent_list
@classmethod
def from_crawler(cls, crawler):
return cls(
user_agent_list=crawler.settings.get('USER_AGENT_LIST')
)
def process_request(self, request, spider):
# 随机选择一个User-Agent
user_agent = random.choice(self.user_agent_list)
if user_agent:
request.headers.setdefault('User-Agent', user_agent)然后实现核心爬虫：
import json
import time
from datetime import datetime
import scrapy
from jd_spider.items import JdProductItem
class JdProductSpider(scrapy.Spider):
name = 'jd_product'
allowed_domains = ['api.jd.com']
# JD API基础URL
base_api_url = 'https://api.jd.com/routerjson'
# 要爬取的商品分类ID列表
category_ids = [1316, 919, 652, 12259] # 示例分类ID
def start_requests(self):
"""开始请求，遍历分类ID"""
for category_id in self.category_ids:
# 每个分类从第一页开始
yield self.build_category_request(category_id, 1)
def build_category_request(self, category_id, page):
"""构建分类商品列表请求"""
params = {
'method': 'jingdong.category.goods.get',
'category_id': category_id,
'page': page,
'page_size': self.settings.get('PAGE_SIZE', 20)
}
return scrapy.Request(
url=self.base_api_url,
meta={
'params': params,
'category_id': category_id,
'page': page
},
callback=self.parse_category
)
def parse_category(self, response):
"""解析分类商品列表响应"""
try:
result = json.loads(response.text)
# 检查API调用是否成功
if 'error_response' in result:
self.logger.error(f"API调用错误: {result['error_response']}")
return
# 提取商品列表数据
category_data = result.get('jingdong_category_goods_get_response', {})
result_data = category_data.get('result', {})
products = result_data.get('data', [])
total_count = result_data.get('total_count', 0)
page = response.meta['page']
page_size = self.settings.get('PAGE_SIZE', 20)
category_id = response.meta['category_id']
self.logger.info(f"分类 {category_id} 第 {page} 页，获取到 {len(products)} 个商品，总计 {total_count} 个商品")
# 处理每个商品
for product in products:
# 构建商品详情请求
yield self.build_product_detail_request(product['sku_id'])
# 分页处理
if page * page_size < total_count:
# 控制请求频率，避免触发API限制
time.sleep(self.settings.get('API_REQUEST_DELAY', 1))
# 发送下一页请求
yield self.build_category_request(category_id, page + 1)
except Exception as e:
self.logger.error(f"解析分类响应出错: {str(e)}，响应内容: {response.text}")
def build_product_detail_request(self, product_id):
"""构建商品详情请求"""
params = {
'method': 'jingdong.product.detail.get',
'sku_id': product_id
}
return scrapy.Request(
url=self.base_api_url,
meta={'params': params, 'product_id': product_id},
callback=self.parse_product_detail
)
def parse_product_detail(self, response):
"""解析商品详情响应"""
try:
result = json.loads(response.text)
# 检查API调用是否成功
if 'error_response' in result:
self.logger.error(f"商品 {response.meta['product_id']} API调用错误: {result['error_response']}")
return
# 提取商品详情数据
product_data = result.get('jingdong_product_detail_get_response', {}).get('result', {})
# 创建商品Item
item = JdProductItem()
# 填充商品信息
item['product_id'] = product_data.get('sku_id')
item['name'] = product_data.get('name')
item['price'] = product_data.get('price')
item['original_price'] = product_data.get('market_price')
item['brand'] = product_data.get('brand_name')
item['stock'] = product_data.get('stock_num')
item['sales_count'] = product_data.get('sales_count')
item['category'] = product_data.get('category_name')
item['category_id'] = product_data.get('category_id')
item['shop_name'] = product_data.get('shop_name')
item['shop_id'] = product_data.get('shop_id')
item['comments_count'] = product_data.get('comment_count')
item['score'] = product_data.get('score')
item['url'] = f"https://item.jd.com/{product_data.get('sku_id')}.html"
item['crawl_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
yield item
except Exception as e:
self.logger.error(f"解析商品 {response.meta['product_id']} 详情出错: {str(e)}，响应内容: {response.text}")最后配置项目设置：
# JD API 配置
JD_APP_KEY = 'your_app_key' # 替换为你的app key
JD_APP_SECRET = 'your_app_secret' # 替换为你的app secret
# 爬虫设置
BOT_NAME = 'jd_spider'
SPIDER_MODULES = ['jd_spider.spiders']
NEWSPIDER_MODULE = 'jd_spider.spiders'
# 并发设置
CONCURRENT_REQUESTS = 4 # 并发请求数
CONCURRENT_REQUESTS_PER_DOMAIN = 2 # 每个域名的并发请求数
# 下载延迟，控制请求频率
DOWNLOAD_DELAY = 2
API_REQUEST_DELAY = 1 # API请求之间的延迟
# 每页商品数量
PAGE_SIZE = 20
# 用户代理列表
USER_AGENT_LIST = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/89.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15'
]
# 下载中间件配置
DOWNLOADER_MIDDLEWARES = {
'jd_spider.middlewares.JdApiAuthMiddleware': 543,
'jd_spider.middlewares.RandomUserAgentMiddleware': 544,
'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': None, # 禁用默认的UserAgent中间件
}
# 管道配置，可配置保存到数据库或文件
ITEM_PIPELINES = {
'jd_spider.pipelines.JdSpiderPipeline': 300,
# 如需保存到CSV，取消下面的注释
# 'scrapy.pipelines.files.CsvItemExporter': 400,
}
# 日志设置
LOG_LEVEL = 'INFO'
LOG_FILE = 'jd_spider.log'
# 遵守robots.txt规则
ROBOTSTXT_OBEY = True
# 自动限速设置
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_START_DELAY = 5
AUTOTHROTTLE_MAX_DELAY = 60
AUTOTHROTTLE_TARGET_CONCURRENCY = 1.0数据存储
实现一个简单的管道，将数据保存到 JSON 文件：
import json
import os
from datetime import datetime
class JdSpiderPipeline:
def __init__(self):
# 创建数据存储目录
self.data_dir = 'data'
if not os.path.exists(self.data_dir):
os.makedirs(self.data_dir)
# 按日期创建数据文件
date_str = datetime.now().strftime('%Y%m%d')
self.filename = os.path.join(self.data_dir, f'jd_products_{date_str}.json')
# 打开文件
self.file = open(self.filename, 'a', encoding='utf-8')
def process_item(self, item, spider):
"""处理每个商品数据"""
# 将Item转换为字典并写入文件
line = json.dumps(dict(item), ensure_ascii=False) + '\n'
self.file.write(line)
return item
def close_spider(self, spider):
"""爬虫关闭时关闭文件"""
self.file.close()
spider.logger.info(f"数据已保存到 {self.filename}")合规性考虑
API 使用规范：严格遵守API 使用条款，不超过请求频率限制
数据用途：确保采集的数据仅用于合法用途，不侵犯用户隐私和平台权益
身份认证：始终使用合法申请的 API 密钥进行访问
请求频率控制：通过设置合理的延迟控制请求频率，避免给服务器造成过大负担
总结与扩展
本文介绍的 Scrapy 集成 JD API 方案，实现了高效且合规的商品数据采集。该方案的优势在于：
合规性高，符合平台规范
稳定性好，API 接口相对稳定，减少维护成本
可扩展性强，可以根据需要扩展采集的字段和范围
未来可以从以下方面进行扩展：
集成代理池，进一步提高采集稳定性
实现分布式爬取，提高大规模数据采集效率
添加数据清洗和分析模块，实现从采集到分析的完整流程
集成数据库存储，支持更复杂的查询和分析需求
通过这种方案，企业和开发者可以在合法合规的前提下，高效获取电商平台数据，为商业决策提供支持。
