在电商数据分析、竞品监控和价格跟踪等场景中，实时获取商品详情页数据具有重要价值。本文将以京东商品详情页为例，介绍如何开发 API 接口来获取并解析商品数据，包括接口设计、数据爬取、解析处理及实战代码实现。
需求分析
京东商品详情页包含丰富的商品信息，我们需要获取的核心数据包括：
商品基本信息（名称、价格、图片、商品 ID 等）
商品规格参数
商品促销信息
库存状态
卖家信息
技术方案
我们将采用以下技术栈实现：
Python 作为开发语言
Requests 库用于发送 HTTP 请求
BeautifulSoup 用于解析 HTML 页面
Flask 框架用于构建 API 服务
JSON 用于数据交换格式
实战开发
1. 分析京东商品详情页结构
京东商品详情页 URL 格式通常为：https://item.jd.com/[商品ID].html，例如https://item.jd.com/100012345678.html。
通过浏览器开发者工具分析页面结构，可以确定各数据在 HTML 中的位置，为解析做准备。
2. 开发商品数据获取工具类
首先实现一个工具类，用于从京东商品详情页爬取并解析数据：
import requests
from bs4 import BeautifulSoup
import json
import re
import time
from fake_useragent import UserAgent
class JDProductCrawler:
def __init__(self):
# 初始化请求头，模拟浏览器访问
self.ua = UserAgent()
self.headers = {
'User-Agent': self.ua.random,
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1'
}
# 设置超时时间
self.timeout = 10
def get_product_detail(self, product_id):
"""
获取商品详情信息
:param product_id: 商品ID
:return: 商品详情字典
"""
try:
# 构建商品详情页URL
url = f'https://item.jd.com/{product_id}.html'
# 发送请求
response = requests.get(url, headers=self.headers, timeout=self.timeout)
response.encoding = 'utf-8'
# 检查请求是否成功
if response.status_code != 200:
return {"error": f"请求失败，状态码: {response.status_code}"}
# 解析HTML
soup = BeautifulSoup(response.text, 'lxml')
# 提取商品数据
product_data = {
'product_id': product_id,
'title': self._extract_title(soup),
'price': self._extract_price(product_id),
'original_price': self._extract_original_price(soup),
'images': self._extract_images(soup),
'brand': self._extract_brand(soup),
'specifications': self._extract_specifications(soup),
'stock_status': self._check_stock(product_id),
'shop_info': self._extract_shop_info(soup),
'crawl_time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
}
return product_data
except Exception as e:
return {"error": f"获取商品信息失败: {str(e)}"}
def _extract_title(self, soup):
"""提取商品标题"""
title_tag = soup.find('div', class_='sku-name')
return title_tag.get_text(strip=True) if title_tag else None
def _extract_price(self, product_id):
"""提取商品当前价格（通过API获取）"""
try:
price_url = f'https://p.3.cn/prices/mgets?skuIds=J_{product_id}'
response = requests.get(price_url, headers=self.headers, timeout=self.timeout)
if response.status_code == 200 and response.text:
price_data = json.loads(response.text)
return price_data[0].get('p') if price_data else None
return None
except Exception as e:
print(f"提取价格失败: {e}")
return None
def _extract_original_price(self, soup):
"""提取商品原价"""
original_price_tag = soup.find('del', class_='p-price')
if original_price_tag:
return original_price_tag.get_text(strip=True)
# 尝试其他位置
price_tags = soup.find_all('span', class_='price')
for tag in price_tags:
if '原价' in tag.get_text():
return tag.get_text().replace('原价：', '').strip()
return None
def _extract_images(self, soup):
"""提取商品图片"""
images = []
# 主图
main_image_tag = soup.find('img', id='spec-img')
if main_image_tag and 'src' in main_image_tag.attrs:
main_image = main_image_tag['src']
if main_image.startswith('//'):
main_image = 'https:' + main_image
images.append(main_image)
# 缩略图
thumb_tags = soup.find_all('img', class_='ps-item')
for tag in thumb_tags:
if 'src' in tag.attrs:
img_url = tag['src']
if img_url.startswith('//'):
img_url = 'https:' + img_url
# 替换缩略图为原图
img_url = img_url.replace('/n9/', '/n1/')
images.append(img_url)
# 去重
return list(set(images))
def _extract_brand(self, soup):
"""提取品牌信息"""
brand_tag = soup.find('ul', class_='parameter2 p-parameter-list')
if brand_tag:
brand_item = brand_tag.find('li', string=re.compile('品牌'))
if brand_item:
return brand_item.get_text().split('：')[-1].strip()
return None
def _extract_specifications(self, soup):
"""提取商品规格参数"""
specs = {}
spec_tag = soup.find('ul', class_='parameter2 p-parameter-list')
if spec_tag:
for li in spec_tag.find_all('li'):
text = li.get_text().strip()
if '：' in text:
key, value = text.split('：', 1)
specs[key.strip()] = value.strip()
return specs
def _check_stock(self, product_id):
"""检查库存状态"""
try:
stock_url = f'https://c0.3.cn/stock?skuId={product_id}&area=1_72_2799_0&venderId=1000000000&cat=1316,1381,1456&buyNum=1'
response = requests.get(stock_url, headers=self.headers, timeout=self.timeout)
if response.status_code == 200 and response.text:
# 解析JSONP响应
stock_data = json.loads(response.text[len('jQuery17209576466463223437_1622505777672('):-1])
stock_status = stock_data.get('stock', {}).get('stockStatus', 0)
return '有货' if stock_status == 1 else '无货'
return '未知'
except Exception as e:
print(f"检查库存失败: {e}")
return '未知'
def _extract_shop_info(self, soup):
"""提取店铺信息"""
shop_info = {}
shop_link = soup.find('a', class_='shopname')
if shop_link:
shop_info['name'] = shop_link.get_text(strip=True)
shop_info['url'] = shop_link['href'] if 'href' in shop_link.attrs else None
if shop_info['url'] and shop_info['url'].startswith('//'):
shop_info['url'] = 'https:' + shop_info['url']
# 检查是否为京东自营
self_run_tag = soup.find('a', class_='u-jd')
shop_info['is_self_operated'] = True if self_run_tag else False
return shop_info3. 开发 API 服务
使用 Flask 框架将上述工具类封装为 API 服务，方便其他系统调用：
from flask import Flask, request, jsonify
from flask_cors import CORS
from jd_product_crawler import JDProductCrawler
import logging
import time
# 初始化Flask应用
app = Flask(__name__)
# 允许跨域请求
CORS(app)
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 初始化爬虫实例
crawler = JDProductCrawler()
# 用于限制请求频率的字典
request_timestamps = {}
def is_request_allowed(ip):
"""检查请求是否允许（限制每分钟最多60次请求）"""
current_time = time.time()
if ip not in request_timestamps:
request_timestamps[ip] = []
# 清除1分钟前的请求记录
request_timestamps[ip] = [t for t in request_timestamps[ip] if current_time - t < 60]
# 检查请求次数
if len(request_timestamps[ip]) < 60:
request_timestamps[ip].append(current_time)
return True
return False
@app.route('/api/jd/product', methods=['GET'])
def get_jd_product():
"""获取京东商品详情API接口"""
# 获取客户端IP
client_ip = request.remote_addr
# 检查请求频率
if not is_request_allowed(client_ip):
logger.warning(f"IP {client_ip} 请求过于频繁")
return jsonify({
"error": "请求过于频繁，请稍后再试",
"code": 429
}), 429
# 获取商品ID参数
product_id = request.args.get('id')
if not product_id or not product_id.isdigit():
return jsonify({
"error": "无效的商品ID，请提供数字形式的商品ID",
"code": 400
}), 400
logger.info(f"IP {client_ip} 请求商品ID: {product_id} 的详情")
# 获取商品详情
try:
product_data = crawler.get_product_detail(product_id)
# 检查是否有错误
if "error" in product_data:
logger.error(f"获取商品 {product_id} 详情失败: {product_data['error']}")
return jsonify({
"error": product_data['error'],
"code": 500
}), 500
# 返回成功响应
return jsonify({
"data": product_data,
"code": 200,
"message": "success"
})
except Exception as e:
logger.error(f"处理商品 {product_id} 请求时发生错误: {str(e)}")
return jsonify({
"error": "服务器内部错误",
"code": 500
}), 500
@app.route('/api/jd/products', methods=['GET'])
def get_jd_products():
"""批量获取京东商品详情API接口"""
# 获取客户端IP
client_ip = request.remote_addr
# 获取商品ID列表
product_ids = request.args.get('ids', '').split(',')
if not product_ids or len(product_ids) > 10: # 限制最多10个商品
return jsonify({
"error": "请提供有效的商品ID列表，最多10个",
"code": 400
}), 400
# 过滤无效ID
valid_ids = [pid for pid in product_ids if pid and pid.isdigit()]
if not valid_ids:
return jsonify({
"error": "没有有效的商品ID",
"code": 400
}), 400
logger.info(f"IP {client_ip} 批量请求商品ID: {valid_ids} 的详情")
# 批量获取商品详情
results = []
for pid in valid_ids:
# 控制请求频率，避免被封
time.sleep(1)
product_data = crawler.get_product_detail(pid)
results.append({
"product_id": pid,
"data": product_data if "error" not in product_data else None,
"error": product_data["error"] if "error" in product_data else None
})
# 返回成功响应
return jsonify({
"data": results,
"code": 200,
"message": "success"
})
@app.route('/health', methods=['GET'])
def health_check():
"""健康检查接口"""
return jsonify({
"status": "healthy",
"code": 200
})
if __name__ == '__main__':
# 启动服务，默认端口5000
app.run(host='0.0.0.0', port=5000, debug=False)4. API 接口使用示例
开发完成后，可以通过以下方式调用 API 接口获取数据：
import requests
import json
def get_single_product(product_id):
"""获取单个商品详情"""
api_url = f'http://localhost:5000/api/jd/product?id={product_id}'
try:
response = requests.get(api_url)
result = response.json()
if result['code'] == 200:
print(f"成功获取商品 {product_id} 详情:")
print(json.dumps(result['data'], ensure_ascii=False, indent=2))
return result['data']
else:
print(f"获取商品 {product_id} 详情失败: {result['error']}")
return None
except Exception as e:
print(f"调用API时发生错误: {str(e)}")
return None
def get_multiple_products(product_ids):
"""批量获取商品详情"""
ids_str = ','.join(product_ids)
api_url = f'http://localhost:5000/api/jd/products?ids={ids_str}'
try:
response = requests.get(api_url)
result = response.json()
if result['code'] == 200:
print(f"成功获取 {len(product_ids)} 个商品详情:")
for item in result['data']:
print(f"商品ID: {item['product_id']}")
if item['error']:
print(f" 错误: {item['error']}")
else:
print(f" 名称: {item['data']['title']}")
print(f" 价格: {item['data']['price']}")
return result['data']
else:
print(f"批量获取商品详情失败: {result['error']}")
return None
except Exception as e:
print(f"调用API时发生错误: {str(e)}")
return None
if __name__ == '__main__':
# 示例：获取单个商品详情
product_id = '100012345678' # 替换为实际的京东商品ID
get_single_product(product_id)
# 示例：批量获取商品详情
product_ids = ['100012345678', '100008348544', '100015382879'] # 替换为实际的京东商品ID列表
get_multiple_products(product_ids)数据解析与应用
获取到的商品数据可以用于多种场景：
价格监控：定期获取商品价格，分析价格波动趋势，设置降价提醒
竞品分析：对比不同商家同类商品的价格、规格和促销策略
库存跟踪：监控热门商品库存状态，及时补货或调整销售策略
数据分析：构建商品数据库，进行市场趋势分析和消费者行为研究
注意事项
法律合规：确保爬虫行为符合网站 robots 协议和相关法律法规，避免过度请求给服务器造成负担
反爬机制：京东有反爬机制，实际使用中可能需要更复杂的 IP 代理池和请求头轮换策略
数据更新：商品信息可能随时变化，需定期更新数据以保证准确性
异常处理：网络请求可能失败，需完善异常处理机制，确保系统稳定性
总结
本文介绍了如何开发 API 接口来实时获取京东商品详情页数据，包括页面分析、数据爬取、解析处理和 API 服务构建。通过这个实战案例，我们可以掌握 API 开发、网页爬取和数据解析的核心技术，为电商数据分析和应用开发提供基础支持。
在实际应用中，还可以根据具体需求扩展功能，如添加缓存机制提高性能、增加更多数据字段、实现更复杂的反爬策略等。
