在当今的电商时代，获取商品的实时数据对于市场分析、竞品监控和价格策略制定至关重要。本文将带您从零开始，开发一个淘宝商品实时数据采集接口，通过这个接口可以获取商品的基本信息、价格、销量等关键数据。
技术选型
淘宝商品数据采集接口，我们将使用以下技术栈：
Python：作为主要开发语言，简洁高效
Requests：处理 HTTP 请求
BeautifulSoup：解析 HTML 页面
Flask：搭建 API 服务
Redis：缓存数据，减轻服务器压力
实现思路
分析淘宝商品页面结构，确定需要采集的数据字段
编写爬虫代码，模拟浏览器请求并解析页面
实现数据缓存机制，避免频繁请求
搭建 API 服务，提供统一的数据访问接口
添加异常处理和反爬措施
代码实现
1. 项目结构
taobao-crawler/ ├── app.py # Flask应用主文件 ├── crawler.py # 爬虫核心逻辑 ├── cache.py # 缓存相关操作 ├── config.py # 配置文件 └── requirements.txt # 依赖包列表
2. 依赖安装
首先创建 requirements.txt 文件：
flask==2.0.1 requests==2.26.0 beautifulsoup4==4.10.0 redis==3.5.3 python-dotenv==0.19.0 fake_useragent==0.1.11
安装依赖：
pip install -r requirements.txt
3. 配置文件
import os
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
# 爬虫配置
TIMEOUT = 10
RETRY_TIMES = 3
CACHE_EXPIRE = 300 # 缓存过期时间，单位秒
# Redis配置
REDIS_HOST = os.getenv('REDIS_HOST', 'localhost')
REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
REDIS_DB = int(os.getenv('REDIS_DB', 0))
# API配置
API_HOST = os.getenv('API_HOST', '0.0.0.0')
API_PORT = int(os.getenv('API_PORT', 5000))4. 缓存模块
import redis import json from config import REDIS_HOST, REDIS_PORT, REDIS_DB, CACHE_EXPIRE # 初始化Redis连接 redis_client = redis.Redis( host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, decode_responses=True ) def get_cache(key): """获取缓存数据""" data = redis_client.get(key) if data: return json.loads(data) return None def set_cache(key, data, expire=CACHE_EXPIRE): """设置缓存数据""" redis_client.setex(key, expire, json.dumps(data)) def delete_cache(key): """删除缓存数据""" redis_client.delete(key)
5. 爬虫核心逻辑
import requests
from bs4 import BeautifulSoup
import re
import json
from fake_useragent import UserAgent
from config import TIMEOUT, RETRY_TIMES
from cache import get_cache, set_cache
# 初始化UserAgent
ua = UserAgent()
def get_taobao_item(item_id):
"""
获取淘宝商品信息
:param item_id: 商品ID
:return: 商品信息字典
"""
# 先从缓存获取
cache_key = f"taobao:item:{item_id}"
cached_data = get_cache(cache_key)
if cached_data:
return cached_data
# 构建商品详情页URL
url = f"https://item.taobao.com/item.htm?id={item_id}"
# 设置请求头，模拟浏览器
headers = {
"User-Agent": 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"
}
# 发送请求，带重试机制
for _ in range(RETRY_TIMES):
try:
response = requests.get(url, headers=headers, timeout=TIMEOUT)
response.encoding = "gbk" # 淘宝页面通常使用gbk编码
if response.status_code == 200:
# 解析页面
data = parse_item_page(response.text, item_id)
if data:
# 存入缓存
set_cache(cache_key, data)
return data
except Exception as e:
print(f"获取商品信息失败: {str(e)}")
return None
def parse_item_page(html, item_id):
"""
解析商品详情页HTML
:param html: 页面HTML内容
:param item_id: 商品ID
:return: 解析后的商品信息
"""
soup = BeautifulSoup(html, 'lxml')
# 提取商品基本信息
result = {
"item_id": item_id,
"title": "",
"price": "",
"sales": 0,
"shop_name": "",
"shop_url": "",
"category": "",
"images": [],
"specifications": {}
}
# 获取标题
title_tag = soup.find('h3', class_='tb-main-title')
if title_tag:
result["title"] = title_tag.get_text(strip=True)
# 获取价格
price_tag = soup.find('em', class_='tb-rmb-num')
if price_tag:
result["price"] = price_tag.get_text(strip=True)
# 获取销量（淘宝商品详情页销量可能通过JS加载，这里采用另一种方式）
sales_script = re.search(r'aucNumId:"(\d+)",.*?viewSales:"(.*?)"', html)
if sales_script and sales_script.group(2):
sales_text = sales_script.group(2)
# 提取数字
sales_num = re.search(r'\d+', sales_text)
if sales_num:
result["sales"] = int(sales_num.group(0))
# 获取店铺信息
shop_name_tag = soup.find('a', class_='s-logo-shopname')
if shop_name_tag:
result["shop_name"] = shop_name_tag.get_text(strip=True)
result["shop_url"] = shop_name_tag.get('href', '')
# 获取商品图片
image_tags = soup.find_all('img', class_='J_ItemImg')
for img in image_tags:
img_url = img.get('src')
if img_url and img_url.startswith('//'):
img_url = 'https:' + img_url
if img_url:
result["images"].append(img_url)
# 获取商品规格信息（简化版）
spec_script = re.search(r'var g_config = (.*?);\n', html)
if spec_script:
try:
spec_data = json.loads(spec_script.group(1))
if "item" in spec_data and "category" in spec_data["item"]:
result["category"] = spec_data["item"]["category"]
except:
pass
return result
def search_taobao(keyword, page=1):
"""
搜索淘宝商品
:param keyword: 搜索关键词
:param page: 页码
:return: 商品列表
"""
# 先从缓存获取
cache_key = f"taobao:search:{keyword}:{page}"
cached_data = get_cache(cache_key)
if cached_data:
return cached_data
# 构建搜索URL
url = f"https://s.taobao.com/search?q={keyword}&s={(page-1)*44}"
# 设置请求头
headers = {
"User-Agent": 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",
"Referer": "https://www.taobao.com/"
}
# 发送请求
try:
response = requests.get(url, headers=headers, timeout=TIMEOUT)
response.encoding = "utf-8"
if response.status_code == 200:
# 解析搜索结果
items = parse_search_result(response.text)
# 存入缓存
set_cache(cache_key, items)
return items
except Exception as e:
print(f"搜索商品失败: {str(e)}")
return []
def parse_search_result(html):
"""解析搜索结果页面"""
# 从脚本中提取商品数据
pattern = re.compile(r'g_page_config = (.*?);\n')
match = pattern.search(html)
if not match:
return []
try:
data = json.loads(match.group(1))
if "mods" in data and "itemlist" in data["mods"] and "data" in data["mods"]["itemlist"]:
items = data["mods"]["itemlist"]["data"]["auctions"]
# 提取需要的字段
result = []
for item in items:
result.append({
"item_id": item.get("nid", ""),
"title": item.get("title", ""),
"price": item.get("view_price", ""),
"sales": item.get("view_sales", ""),
"location": item.get("item_loc", ""),
"shop_name": item.get("nick", ""),
"shop_url": item.get("shopLink", ""),
"pic_url": item.get("pic_url", "").replace("//", "https://") if item.get("pic_url") else ""
})
return result
except Exception as e:
print(f"解析搜索结果失败: {str(e)}")
return []6. API 服务实现
from flask import Flask, request, jsonify
from crawler import get_taobao_item, search_taobao
from config import API_HOST, API_PORT
app = Flask(__name__)
@app.route('/api/item/<item_id>', methods=['GET'])
def get_item(item_id):
"""获取单个商品详情"""
try:
data = get_taobao_item(item_id)
if data:
return jsonify({
"code": 200,
"message": "success",
"data": data
})
else:
return jsonify({
"code": 404,
"message": "商品不存在或获取失败",
"data": None
})
except Exception as e:
return jsonify({
"code": 500,
"message": f"服务器错误: {str(e)}",
"data": None
})
@app.route('/api/search', methods=['GET'])
def search():
"""搜索商品"""
try:
keyword = request.args.get('keyword', '')
page = int(request.args.get('page', 1))
if not keyword:
return jsonify({
"code": 400,
"message": "请提供搜索关键词",
"data": None
})
data = search_taobao(keyword, page)
return jsonify({
"code": 200,
"message": "success",
"data": {
"items": data,
"page": page,
"count": len(data)
}
})
except Exception as e:
return jsonify({
"code": 500,
"message": f"服务器错误: {str(e)}",
"data": None
})
@app.route('/api/health', methods=['GET'])
def health_check():
"""健康检查接口"""
return jsonify({
"code": 200,
"message": "service is running",
"timestamp": int(request.args.get('timestamp', 0)) or None
})
if __name__ == '__main__':
app.run(host=API_HOST, port=API_PORT, debug=True)接口使用示例
1. 获取单个商品详情
请求：
GET /api/item/586722645217
响应示例：
{
"code": 200,
"message": "success",
"data": {
"item_id": "586722645217",
"title": "示例商品标题",
"price": "199.00",
"sales": 1256,
"shop_name": "示例店铺",
"shop_url": "https://shop12345678.taobao.com",
"category": "服饰鞋包 > 女装",
"images": [
"https://img.alicdn.com/imgextra/i1/abc.jpg",
"https://img.alicdn.com/imgextra/i2/def.jpg"
],
"specifications": {}
}
}反爬措施与优化
User-Agent 随机化：使用 fake_useragent 库生成不同的浏览器标识，避免被识别为爬虫
请求频率控制：可以在实际应用中添加请求间隔控制
IP 代理池：对于大规模采集，可以使用 IP 代理池避免 IP 被封禁
数据缓存：使用 Redis 缓存减轻目标服务器压力，同时提高接口响应速度
异常处理：完善的异常处理机制，提高程序稳定性
注意事项
本接口仅用于学习和研究目的，使用时请遵守淘宝的 robots 协议和相关规定
频繁的请求可能会导致 IP 被封禁，建议合理控制请求频率
淘宝页面结构可能会发生变化，需要定期维护爬虫代码
商业用途请联系淘宝官方获取合法的数据接口
通过本文的实战教程，我们实现了一个简单但功能完整的淘宝商品数据采集接口。您可以根据实际需求扩展更多功能，如批量采集、数据导出、定时更新等。在实际应用中，还需要考虑更多的反爬策略和性能优化，以确保接口的稳定运行。
