分析中 已回复 待规划 {{opt.name}}

分析中 已回复 待规划

Python 爬虫获取淘宝商品详情（标题、主图、SKU、价格）实战指南

一、前言：为什么需要这篇文章

淘宝商品详情数据（标题、主图、SKU、价格）是电商数据分析、竞品监控、选品工具开发的核心资源。获取这些数据主要有两条路径：官方 API（推荐） 和 网页爬虫（补充）。本文将系统介绍两种方案的技术实现、反爬对抗策略，以及至关重要的合规边界。

二、方案一：淘宝开放平台官方 API（首选）

淘宝开放平台（TOP）提供官方 API 接口，合法、稳定、数据完整，是商业项目的首选方案。

2.1 前置准备

1. 注册开发者账号：访问 淘宝开放平台，完成实名认证（企业账号权限更高）
2. 创建应用：获取 AppKey 和 AppSecret
3. 申请权限：申请 taobao.item.get 接口权限，说明业务场景（如"商品库存自动同步"）
4. OAuth 授权：商家授权后获取 Access Token

2.2 核心接口：taobao.item.get

该接口可获取商品标题、价格、SKU、库存、主图等完整信息。

关键参数说明：

2.3 Python 调用示例

import hashlib
import time
import urllib.parse
import requests

def generate_taobao_sign(params, app_secret):
    """
    生成淘宝 API 签名（MD5/HMAC-MD5）
    签名错误会直接返回 400，这是最常见的坑
    """
    # 1. 排除 sign 参数，按参数名 ASCII 升序排序
    sorted_params = sorted([(k, v) for k, v in params.items() if k != "sign"])
    # 2. 拼接为 "key=value&key=value" 格式
    sign_str = "&".join([f"{k}={urllib.parse.quote_plus(str(v))}" for k, v in sorted_params])
    # 3. 末尾拼接 AppSecret，MD5 加密后转大写
    sign_str += app_secret
    return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()

def get_taobao_item_detail(item_id, app_key, app_secret, access_token=None):
    """
    获取淘宝商品详情（官方 API）
    """
    # 1. 构造请求参数
    params = {
        "app_key": app_key,
        "method": "taobao.item.get",
        "format": "json",
        "v": "2.0",
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "num_iid": item_id,
        "fields": "num_iid,title,price,stock,pic_url,sku,props_name,item_imgs,prop_imgs,ai_tag"
    }
    
    # 2. 生成签名
    params["sign"] = generate_taobao_sign(params, app_secret)
    
    # 3. 发送请求（淘宝 API 固定域名）
    url = "https://eco.taobao.com/router/rest"
    try:
        response = requests.get(url, params=params, timeout=10)
        result = response.json()
        
        # 4. 结果解析
        if "error_response" in result:
            error_msg = result["error_response"]["msg"]
            raise Exception(f"接口调用失败：{error_msg}")
        
        item = result["item_get_response"]["item"]
        return {
            "title": item.get("title"),
            "price": item.get("price"),
            "pic_url": item.get("pic_url"),  # 主图
            "item_imgs": item.get("item_imgs", {}).get("item_img", []),  # 详情图列表
            "sku": item.get("skus", {}).get("sku", []),  # SKU 列表
            "props_name": item.get("props_name"),  # 属性规格
            "ai_tag": item.get("ai_tag", "无")
        }
        
    except Exception as e:
        print(f"请求异常: {e}")
        return None

# 调用示例
if __name__ == "__main__":
    APP_KEY = "你的AppKey"
    APP_SECRET = "你的AppSecret"
    ITEM_ID = "123456789012"
    
    data = get_taobao_item_detail(ITEM_ID, APP_KEY, APP_SECRET)
    if data:
        print(f"商品标题：{data['title']}")
        print(f"商品价格：{data['price']}")
        print(f"AI 标签：{data['ai_tag']}")
        
        使用官方 SDK 更简便：


bash复制

pip install top-sdk-python



Python复制

from top.api import TopClient
from top.request import TaobaoItemGetRequest

client = TopClient(
    appkey="你的AppKey",
    secret="你的AppSecret",
    gateway="https://gw.api.taobao.com/router/rest"
)

req = TaobaoItemGetRequest()
req.num_iid = "123456789"
req.fields = "num_iid,title,price,pic_url,sku"

resp = client.execute(req, session="你的AccessToken")
print(resp)

2.4 返回数据结构解析

{
  "item_get_response": {
    "item": {
      "num_iid": "123456789",
      "title": "2026新款夏季连衣裙",
      "price": "199.00",
      "pic_url": "https://img.alicdn.com/...jpg",
      "item_imgs": {
        "item_img": [
          {"url": "https://img.alicdn.com/...1.jpg"},
          {"url": "https://img.alicdn.com/...2.jpg"}
        ]
      },
      "skus": {
        "sku": [
          {
            "sku_id": "123456",
            "price": "199.00",
            "properties": "1627207:28341;20509:28316",
            "properties_name": "颜色:红色;尺码:M",
            "quantity": "100"
          }
        ]
      },
      "props_name": "颜色:红色,蓝色;尺码:S,M,L"
    }
  }
}

三、方案二：网页爬虫（补充方案）

当官方 API 无法满足需求时（如需要抓取未授权商品、历史价格等），可采用爬虫方案。但必须严格遵守合规边界。

3.1 淘宝反爬机制分析

淘宝采用多层防护体系：

3.2 合规爬虫实现

核心原则：只爬取公开可见数据，不破解任何技术措施，控制频率避免影响服务器。

import requests
import time
import random
import re
import json
from urllib.parse import urlencode
from fake_useragent import UserAgent

class TaobaoItemSpider:
    def __init__(self):
        self.ua = UserAgent()
        self.session = requests.Session()
        self.session.headers.update({
            '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.9,en;q=0.8',
            'Accept-Encoding': 'gzip, deflate, br',
            'Connection': 'keep-alive',
        })
        # 代理池（用于分散请求，非伪造身份）
        self.proxies = []
        self.current_proxy_index = 0
        
    def get_proxy(self):
        """轮询获取代理"""
        if not self.proxies:
            return None
        proxy = self.proxies[self.current_proxy_index]
        self.current_proxy_index = (self.current_proxy_index + 1) % len(self.proxies)
        return {"http": proxy, "https": proxy}
    
    def fetch_item_page(self, item_id, max_retries=3):
        """
        获取商品详情页 HTML
        注意：淘宝商品页结构经常变化，需定期维护选择器
        """
        url = f"https://item.taobao.com/item.htm?id={item_id}"
        
        for attempt in range(max_retries):
            try:
                # 关键：频率控制（合规底线）
                time.sleep(random.uniform(5, 10))
                
                proxy = self.get_proxy()
                response = self.session.get(
                    url, 
                    proxies=proxy, 
                    timeout=15,
                    allow_redirects=True
                )
                
                if response.status_code == 200:
                    return response.text
                elif response.status_code == 403:
                    print(f"IP 被封禁，等待 60 秒后重试...")
                    time.sleep(60)
                    continue
                elif "验证码" in response.text or "security" in response.text:
                    print("触发验证码，停止爬取")
                    return None
                    
            except Exception as e:
                print(f"请求失败: {e}")
                time.sleep(30)
        
        return None
    
    def extract_item_data(self, html):
        """
        从 HTML 中提取商品数据
        注意：仅提取明文可见数据，不破解字体反爬
        """
        data = {}
        
        # 提取标题
        title_match = re.search(r'<h1[^>]*class="tb-detail-hd"[^>]*>.*?<a[^>]*>(.*?)</a>', html, re.DOTALL)
        if title_match:
            data['title'] = re.sub(r'<[^>]+>', '', title_match.group(1)).strip()
        
        # 提取主图（从 JSON 数据中提取）
        # 淘宝商品数据通常嵌入在 <script> 标签中
        g_config_match = re.search(r'g_config\s*=\s*({.*?});', html, re.DOTALL)
        if g_config_match:
            try:
                config = json.loads(g_config_match.group(1))
                # 提取主图 URL
                if 'id' in config:
                    data['item_id'] = config['id']
            except:
                pass
        
        # 提取价格（从页面脚本中提取）
        price_match = re.search(r'"defaultItemPrice":"([^"]+)"', html)
        if price_match:
            data['price'] = price_match.group(1)
        
        # 提取 SKU 信息（从 g_config 或 API 接口）
        sku_match = re.search(r'"skuList":(\[.*?\])', html)
        if sku_match:
            try:
                data['sku_list'] = json.loads(sku_match.group(1))
            except:
                pass
        
        return data
    
    def get_item_detail(self, item_id):
        """主入口：获取商品详情"""
        html = self.fetch_item_page(item_id)
        if not html:
            return None
        
        return self.extract_item_data(html)

# 使用示例
if __name__ == "__main__":
    spider = TaobaoItemSpider()
    item_id = "123456789"
    result = spider.get_item_detail(item_id)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    

3.3 移动端 H5 接口（更轻量）

淘宝 H5 页面接口相对简单，返回 JSON 数据：

def get_item_h5(item_id):
    """
    通过淘宝 H5 接口获取商品数据
    接口相对稳定，但仍有反爬限制
    """
    url = "https://h5api.m.taobao.com/h5/mtop.taobao.detail.getdetail/6.0/"
    params = {
        "jsv": "2.4.8",
        "appKey": "12574478",
        "t": str(int(time.time() * 1000)),
        "api": "mtop.taobao.detail.getdetail",
        "v": "6.0",
        "type": "jsonp",
        "dataType": "jsonp",
        "data": json.dumps({"itemNumId": item_id})
    }
    
    try:
        response = requests.get(url, params=params, timeout=10)
        # 解析 JSONP 格式
        json_str = re.search(r'\((.*)\)', response.text).group(1)
        data = json.loads(json_str)
        
        if data.get("ret", [""])[0] == "SUCCESS::调用成功":
            item = data["data"]["item"]
            return {
                "title": item.get("title"),
                "price": item.get("price"),
                "images": item.get("images", []),
                "sku_base": data["data"].get("skuBase", {})
            }
    except Exception as e:
        print(f"H5 接口失败: {e}")
    
    return None
    

四、数据解析：SKU 与主图处理

4.1 SKU 数据解析

SKU（Stock Keeping Unit）是商品规格组合，解析逻辑如下：

def parse_sku_data(api_response):
    """
    解析 SKU 数据，建立规格与价格的映射关系
    """
    sku_list = api_response.get("skus", {}).get("sku", [])
    prop_imgs = api_response.get("prop_imgs", {}).get("prop_img", [])
    
    # 建立属性图片映射
    prop_img_map = {}
    for img in prop_imgs:
        prop_img_map[img.get("properties")] = img.get("url")
    
    parsed_skus = []
    for sku in sku_list:
        sku_id = sku.get("sku_id")
        price = sku.get("price")
        orginal_price = sku.get("orginal_price")
        quantity = sku.get("quantity")
        properties = sku.get("properties")
        properties_name = sku.get("properties_name")  # 如 "颜色:红色;尺码:M"
        
        # 解析属性名
        props = {}
        if properties_name:
            for prop in properties_name.split(";"):
                if ":" in prop:
                    k, v = prop.split(":", 1)
                    props[k] = v
        
        parsed_skus.append({
            "sku_id": sku_id,
            "price": price,
            "original_price": orginal_price,
            "stock": quantity,
            "properties": props,
            "prop_img": prop_img_map.get(properties)
        })
    
    return parsed_skus
    

4.2 主图与详情图处理

def process_images(item_data):
    """
    处理商品图片数据
    """
    # 主图（第一张）
    main_image = item_data.get("pic_url")
    
    # 详情图列表
    item_imgs = item_data.get("item_imgs", {}).get("item_img", [])
    detail_images = [img.get("url") for img in item_imgs]
    
    # SKU 属性图
    prop_imgs = item_data.get("prop_imgs", {}).get("prop_img", [])
    sku_images = {img.get("properties"): img.get("url") for img in prop_imgs}
    
    return {
        "main_image": main_image,
        "detail_images": detail_images,
        "sku_images": sku_images
    }
    

五、API 封装：将能力服务化

使用 Flask 将爬取能力封装为 RESTful API：

from flask import Flask, request, jsonify
from flask_limiter import Limiter
from functools import wraps

app = Flask(__name__)

# 速率限制：防止 API 被滥用
limiter = Limiter(
    app=app,
    key_func=lambda: request.headers.get("X-API-Key", "anonymous"),
    default_limits=["100 per hour"]
)

# 简单的 API Key 认证
VALID_API_KEYS = {"your-api-key-1", "your-api-key-2"}

def require_api_key(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        api_key = request.headers.get("X-API-Key")
        if api_key not in VALID_API_KEYS:
            return jsonify({"error": "Unauthorized"}), 401
        return f(*args, **kwargs)
    return decorated

@app.route("/api/v1/item/<item_id>", methods=["GET"])
@limiter.limit("10 per minute")
@require_api_key
def get_item(item_id):
    """
    获取商品详情 API
    优先使用官方 API，失败时降级到爬虫
    """
    # 尝试官方 API
    data = get_taobao_item_detail(item_id, APP_KEY, APP_SECRET)
    
    if not data:
        # 降级到爬虫（合规前提下）
        spider = TaobaoItemSpider()
        data = spider.get_item_detail(item_id)
    
    if data:
        return jsonify({
            "code": 200,
            "data": data,
            "source": "official_api" if "ai_tag" in data else "spider"
        })
    else:
        return jsonify({"code": 404, "message": "Item not found"}), 404

@app.route("/api/v1/batch/items", methods=["POST"])
@limiter.limit("5 per minute")
@require_api_key
def batch_get_items():
    """
    批量获取商品详情
    """
    item_ids = request.json.get("item_ids", [])
    if len(item_ids) > 20:
        return jsonify({"error": "Max 20 items per batch"}), 400
    
    results = []
    for item_id in item_ids:
        data = get_taobao_item_detail(item_id, APP_KEY, APP_SECRET)
        results.append({
            "item_id": item_id,
            "data": data,
            "success": data is not None
        })
        time.sleep(1)  # 批量请求间隔
    
    return jsonify({"code": 200, "results": results})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)
    

六、合规边界：技术人员的法律红线

6.1 法律风险全景

根据 2025-2026 年最新司法判例，爬虫行为的合法性取决于三个维度：

6.2 2025-2026 年典型判例

1. 招聘平台数据泄露案：程序员日均抓取 50 万条数据并转售，获利 120 万元，判处有期徒刑 3 年 6 个月
2. 电商价格监控案：爬虫每秒 10 次访问致服务器宕机，赔偿 87 万元
3. 地图数据爬取案：爬取 8 亿条地理坐标数据，判处有期徒刑 3 年至 6 个月不等
4. 淘宝"青虎浏览器"案：爬取 6700 万条商品数据，因"实质性替代"被行政处罚

6.3 合规检查清单

class ComplianceChecker:
    """合规性检查工具"""
    
    @staticmethod
    def check_frequency(request_count, time_window_seconds):
        """检查请求频率"""
        max_qps = 0.1  # 每秒不超过 0.1 次
        current_qps = request_count / time_window_seconds
        return current_qps <= max_qps
    
    @staticmethod
    def is_public_data(data_type):
        """检查数据类型是否公开"""
        public_types = {
            'title', 'price', 'pic_url', 'sku', 
            'sales', 'description', 'shop_name'
        }
        return data_type in public_types
    
    @staticmethod
    def check_robots_txt(url, path):
        """检查 robots.txt 协议"""
        # 实现 robots.txt 解析逻辑
        # 淘宝 robots.txt 禁止 /cart/、/trade/ 等路径
        forbidden_paths = ['/cart/', '/trade/', '/buy/']
        return not any(path.startswith(fp) for fp in forbidden_paths)
        

七、部署架构与监控

7.1 推荐架构

[客户端]
    ↓ (HTTPS + API Key)
[Nginx 网关] (限流、SSL、日志)
    ↓
[Flask API 服务] (多进程 Gunicorn)
    ↓ (优先)
[淘宝官方 API]
    ↓ (降级)
[合规爬虫模块]
    ↓
[代理 IP 池] (分散请求)

7.2 监控指标

# 关键监控指标
MONITOR_CONFIG = {
    "max_daily_requests": 10000,      # 日请求上限
    "max_error_rate": 0.05,           # 错误率阈值
    "max_ban_count": 5,               # IP 封禁警告线
    "avg_response_time": 2.0,         # 平均响应时间(秒)
    "compliance_check_interval": 3600 # 合规检查周期(秒)
}

八、总结与建议

8.1 方案对比

如遇任何疑问或有进一步的需求，请随时与我私信或者评论联系。