淘宝商品搜索爬虫 —— 教学设计文档

📚 文档目标

本文档面向爬虫进阶学习者，详细讲解如何实现一个完整的淘宝商品搜索爬虫，突破淘宝 H5 接口的反爬机制（签名、Token、Cookie 认证）。通过本案例，你将学会：

• 如何从浏览器导出的 Netscape 格式 cookies.txt 加载 Cookie
• 如何提取淘宝 H5 接口所需的动态 Token（_m_h5_tk）
• 如何构造 mtop 接口的签名（sign）
• 如何构建搜索请求参数并处理 JSONP 响应
• 如何实现分页爬取并保存为 CSV

本脚本基于淘宝 H5 移动端 API（h5api.m.taobao.com），相比网页端更稳定、数据格式更规范。源码见底部

---

一、整体设计思路

1.1 需求分析

我们要实现一个命令行爬虫，能够：

• 读取用户提供的 Netscape 格式 cookies.txt（通过浏览器插件导出）
• 提取所需的认证信息（_m_h5_tk）
• 构造带签名的 mtop 请求，搜索商品
• 支持分页爬取（每页 48 个商品）
• 提取商品标题、图片、价格、销量、店铺名等关键字段
• 保存为 CSV 文件

1.2 设计原则

• 模拟合法请求：淘宝 H5 接口需要签名（sign）和时间戳（t），签名算法为 md5(token + timestamp + appKey + data)
• Cookie 持久化：使用 http.cookiejar.MozillaCookieJar 直接加载浏览器导出的 cookies.txt，无需手动解析
• 健壮性：网络请求失败重试、JSONP 解析异常处理、字段缺失默认值
• 可配置：关键词、页数、请求间隔、输出路径均通过常量配置

1.3 模块划分

┌─────────────────┐
│   Cookie 加载   │  (MozillaCookieJar)
└────────┬────────┘
         ▼
┌─────────────────┐
│  Token 提取     │  (正则提取 _m_h5_tk)
└────────┬────────┘
         ▼
┌─────────────────┐
│  参数构造       │  (build_search_params)
└────────┬────────┘
         ▼
┌─────────────────┐
│  签名生成       │  (md5)
└────────┬────────┘
         ▼
┌─────────────────┐
│  请求 & 解析    │  (JSONP)
└────────┬────────┘
         ▼
┌─────────────────┐
│  CSV 存储       │
└─────────────────┘

---

二、核心模块详解

2.1 Cookie 加载（Netscape 格式）

淘宝 H5 接口需要完整的登录 Cookie，包括 _m_h5_tk、cookie2、t 等关键字段。使用浏览器插件（如 EditThisCookie）导出为 Netscape 格式的 cookies.txt。

python
def load_cookies_from_netscape(file_path: str) -> requests.Session:
    session = requests.Session()
    cookie_jar = MozillaCookieJar(file_path)
    cookie_jar.load(ignore_discard=True, ignore_expires=True)
    session.cookies.update(cookie_jar)
    return session

教学要点：

• MozillaCookieJar 是 Python 标准库 http.cookiejar 提供的类，专门用于读写 Netscape 格式的 cookie 文件。
• ignore_discard=True 保留会话级 Cookie，ignore_expires=True 忽略过期时间（实际过期后仍会被淘宝拒绝，但加载时不会报错）。
• 将加载的 Cookie 更新到 requests.Session 中，后续所有请求自动携带。

2.2 提取 Token（_m_h5_tk）

淘宝 mtop 接口签名所需的 token 存储在 Cookie 的 _m_h5_tk 字段中，格式为 token_timestamp，我们只需要下划线前面的部分。

python
def extract_token_from_session(session: requests.Session) -> str:
    for cookie in session.cookies:
        if cookie.name == '_m_h5_tk':
            value = cookie.value
            match = re.match(r'([a-f0-9]+)_', value)
            if match:
                return match.group(1)
            raise ValueError(f"格式异常: {value}")
    raise ValueError("未找到 _m_h5_tk Cookie")

设计亮点：

• 遍历 Session 中的 Cookie 对象，而不是手动解析文件，避免重复读取。
• 使用正则提取 token（32 位十六进制字符串），忽略后面的时间戳。

教学要点：

• Cookie 中可能包含多个字段，需要精确匹配名称。
• 淘宝 token 的生成逻辑是服务器端下发，客户端只需原样使用。

2.3 构造搜索参数

淘宝 H5 搜索接口的参数非常多，但大部分可以固定。核心参数：

• q：搜索关键词
• page：页码
• n：每页数量（通常 48）
• sort：排序方式（_coefp 为综合排序）
• 其他如设备信息、网络类型等可以伪造或使用默认值。

python
def build_search_params(keyword: str, page: int) -> Dict:
    params = {
        "q": keyword,
        "page": page,
        "n": 48,
        "sort": "_coefp",
        # ... 其他固定参数
    }
    # 移除值为 None 的键（避免 JSON 中出现 null）
    params = {k: v for k, v in params.items() if v is not None}
    return params

教学要点：

• 参数可以通过浏览器抓包获得（F12 → Network → 搜索请求 → 查看 Payload）。
• 部分参数（如 device、network）可以固定，淘宝不严格校验。
• 移除 None 值，因为 JSON 编码时 null 可能导致签名不一致。

2.4 生成签名（sign）

淘宝 mtop 接口的签名算法（参考官方 H5 页面 JS 代码）：

sign = md5( token + "&" + timestamp + "&" + appKey + "&" + data )

其中：

• token 为 _m_h5_tk 的下划线前部分
• timestamp 为毫秒级时间戳
• appKey 固定为 "12574478"
• data 为请求数据体（包含 appId 和 params）的紧凑 JSON 字符串

python
def build_signature(em_token: str, timestamp: str, app_key: str, data_dict: Dict) -> tuple:
    data_str = json.dumps(data_dict, separators=(',', ':'))
    sign_str = f"{em_token}&{timestamp}&{app_key}&{data_str}"
    sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest()
    return sign, data_str

教学要点：

• separators=(',', ':') 去除 JSON 中的空格，保证签名与淘宝服务器端一致。
• 签名使用的数据体是 整个请求的 data 字段，而不是内层的 params。
• 时间戳必须与签名时使用的一致，服务端会校验时间差。

2.5 构造请求 URL 与参数

淘宝 H5 搜索接口为：

https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/

需要传递的查询参数（Query String）包括：

• jsv、appKey、t（时间戳）、sign（签名）、api、v、type、dataType、callback（JSONP 回调函数名）、data（URL 编码后的数据字符串）

python
query_params = {
    'jsv': '2.7.4',
    'appKey': app_key,
    't': timestamp,
    'sign': sign,
    'api': 'mtop.relationrecommend.wirelessrecommend.recommend',
    'v': '2.0',
    'type': 'jsonp',
    'dataType': 'jsonp',
    'callback': 'mtopjsonp6',          # 固定回调名
    'data': data_str                    # 已经是 JSON 字符串，requests 会自动 URL 编码
}

教学要点：

• 淘宝接口返回 JSONP 格式（包裹在回调函数中），需要正则提取。
• 参数 callback 可以任意，但必须与响应中的函数名一致（通常固定为 mtopjsonp6）。

2.6 发送请求与解析 JSONP

使用 session.get 发送请求，处理 JSONP 响应：

python
resp = session.get(url, params=query_params, headers=headers, timeout=10)
text = resp.text
jsonp_match = re.search(r'mtopjsonp\d+\((.*)\)', text)
if jsonp_match:
    data = json.loads(jsonp_match.group(1))

设计亮点：

• 使用正则匹配 JSONP 回调函数内的 JSON 字符串。
• 若匹配失败，打印前 200 字符便于调试。

2.7 数据提取与清洗

接口返回的 itemsArray 每个元素包含：

• title：商品标题（包含 <span class=H> 高亮标签，需去除）
• pic_path：图片 URL（可能是相对路径，需拼接 https:）
• price：价格字符串（如 "99.00"）
• procity：发货地
• realSales：实际销量（字符串）
• nick：店铺名称

python
row = {
    'title': item.get('title', '').replace('<span class=H>', '').replace('</span>', ''),
    'img': item.get('pic_path', ''),
    'price': item.get('price', ''),
    'procity': item.get('procity', ''),
    'realSales': item.get('realSales', ''),
    'shopName': item.get('nick', ''),
}

教学要点：

• 标题中的高亮标签是搜索结果的关键词高亮，移除后得到纯净标题。
• 图片 URL 若以 // 开头，需补充协议头 https:。

2.8 分页控制与反爬策略

python
for page in range(1, MAX_PAGE + 1):
    items = fetch_page(session, KEYWORD, page, em_token)
    if items is None or not items:
        break
    all_items.extend(items)
    time.sleep(REQUEST_INTERVAL)   # 请求间隔，避免被封

教学要点：

• 每页请求后休眠 REQUEST_INTERVAL 秒，降低请求频率。
• 若某页返回空数据或失败，停止后续翻页（避免无效请求）。
• 可根据需要增加随机延时（random.uniform(0.5, 1.5)）进一步模拟人类行为。

2.9 CSV 保存

使用 csv.DictWriter 保存数据，编码为 utf-8-sig 以便 Excel 直接打开。

python
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(rows)

---

三、异常处理与健壮性设计

场景	处理方式
Cookie 文件不存在或格式错误	抛出 RuntimeError，程序退出并提示用户
_m_h5_tk Cookie 缺失	抛出 ValueError，提示检查 Cookie
网络请求超时或状态码非 200	捕获异常，打印错误并返回 None，上层停止翻页
JSONP 解析失败	打印响应片段，返回 None
接口返回 ret 不为 SUCCESS::调用成功	打印错误信息，返回 None
某个字段缺失	使用 .get(key, default) 提供默认值

设计亮点：

• 失败时不会抛出未捕获异常导致程序崩溃，而是记录日志并优雅退出。
• 分页过程中某页失败立即停止，避免浪费请求。

---

四、使用说明

4.1 环境准备

• Python 3.6+
• 安装依赖：pip install requests

4.2 获取 Cookie 文件

1. 安装浏览器插件（如 Chrome 的 "EditThisCookie" 或 "Get cookies.txt"）。
2. 登录淘宝账号，访问 https://s.taobao.com。
3. 使用插件导出 Cookie 为 Netscape 格式，保存为 cookies.txt 放在脚本目录。

4.3 配置参数

在脚本开头的配置区域修改：

python
KEYWORD = "手机"           # 搜索关键词
MAX_PAGE = 3              # 爬取页数
REQUEST_INTERVAL = 1      # 请求间隔（秒）
OUTPUT_CSV = "taobao.csv" # 输出文件

4.4 运行

bash
python tb.py

4.5 输出示例

title	img	price	procity	realSales	shopName
华为Mate60 Pro 5G手机	//img.alicdn.com/...	5999.00	广东深圳	2.5万+	华为官方旗舰店

---

五、常见问题与解决方案

5.1 提示 加载 Cookie 文件失败

• 原因：cookies.txt 不存在或不是 Netscape 格式。
• 解决：重新导出正确格式的 Cookie 文件。

5.2 提示 未找到 _m_h5_tk Cookie

• 原因：导出的 Cookie 不完整或已过期。
• 解决：重新登录淘宝并导出最新 Cookie。

5.3 请求返回 ret: ['FAIL_SYS_TOKEN_EMPTY::令牌为空']

• 原因：_m_h5_tk 提取错误或签名计算有误。
• 解决：检查 extract_token_from_session 是否正确提取了 token（应为 32 位十六进制字符串）。

5.4 请求返回 ret: ['FAIL_SYS_ILLEGAL_ACCESS::非法访问']

• 原因：时间戳与服务器时间相差过大，或签名算法错误。
• 解决：确保系统时间准确，检查 build_signature 中拼接字符串的顺序和内容是否与淘宝一致。

5.5 某页商品数量少于 48 个

• 正常现象，最后一页可能不足 48 个。脚本会继续爬取下一页直到无数据。

---

六、扩展建议

基于本脚本可以进一步扩展的功能：

• 多关键词循环爬取：读取关键词列表，依次搜索并保存到不同文件。
• 商品详情爬取：获取每个商品的详情页 URL，进一步抓取 SKU、评价等。
• 价格监控：定时运行脚本，记录价格变化并发送通知。
• 代理 IP 池：防止单 IP 请求频率过高被封。
• 异步爬取：使用 aiohttp 提高爬取速度。

---

七、总结

本爬虫完整演示了淘宝 H5 接口的调用流程，包括：

• Cookie 的加载与使用
• Token 提取
• 签名生成（md5）
• 请求参数构造
• JSONP 响应解析
• 分页与反爬策略

通过学习本案例，你将掌握如何破解大多数移动端 API 的签名验证机制，并能够独立开发类似电商平台的爬虫。

下一步练习：尝试修改脚本，增加“价格排序”功能（修改 sort 参数为 _price 或 _sales），并观察请求响应的变化。

完整源码

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
淘宝商品搜索爬虫（使用 Netscape 格式的 cookies.txt）
使用方法：
    1. 将浏览器导出的 cookies.txt 放在脚本同目录下（或修改 COOKIE_FILE 路径）
    2. 运行脚本：python tb.py
    3. 默认搜索关键词“手机”，爬取前 3 页，结果保存到 taobao.csv
"""

import csv
import hashlib
import json
import re
import time
from http.cookiejar import MozillaCookieJar
from typing import Dict, List, Optional

import requests

# ======================== 配置区域 ========================
COOKIE_FILE = "cookies.txt"          # Netscape 格式的 cookie 文件路径
KEYWORD = "手机"                      # 搜索关键词
MAX_PAGE = 3                         # 要爬取的页数（每页 48 个商品）
REQUEST_INTERVAL = 1                 # 请求间隔（秒）
OUTPUT_CSV = "../data/taobao.csv"            # 输出 CSV 文件名

# ======================== 辅助函数 ========================
def load_cookies_from_netscape(file_path: str) -> requests.Session:
    """从 Netscape 格式的 cookies.txt 加载 Cookie 并返回配置好的 Session"""
    session = requests.Session()
    cookie_jar = MozillaCookieJar(file_path)
    try:
        cookie_jar.load(ignore_discard=True, ignore_expires=True)
        session.cookies.update(cookie_jar)
        print(f"成功从 {file_path} 加载了 {len(cookie_jar)} 个 Cookie")
    except Exception as e:
        raise RuntimeError(f"加载 Cookie 文件失败: {e}")
    return session


def extract_token_from_session(session: requests.Session) -> str:
    """从 Session 的 Cookie 中提取 _m_h5_tk 的 token（下划线前部分）"""
    for cookie in session.cookies:
        if cookie.name == '_m_h5_tk':
            value = cookie.value
            match = re.match(r'([a-f0-9]+)_', value)
            if match:
                return match.group(1)
            else:
                raise ValueError(f"_m_h5_tk 值格式异常: {value}")
    raise ValueError("Session 中未找到 _m_h5_tk Cookie，请检查 cookies.txt 是否包含该字段")


def build_signature(em_token: str, timestamp: str, app_key: str, data_dict: Dict) -> tuple:
    """生成 mtop 接口所需的 sign 和 data 字符串，返回 (sign, data_str)"""
    data_str = json.dumps(data_dict, separators=(',', ':'))
    sign_str = f"{em_token}&{timestamp}&{app_key}&{data_str}"
    sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest()
    return sign, data_str


def build_search_params(keyword: str, page: int) -> Dict:
    """构造 data.params 中的搜索参数（参考淘宝 H5 实际请求）"""
    params = {
        "device": "HMA-AL00",
        "isBeta": "false",
        "grayHair": "false",
        "from": "nt_history",
        "brand": "HUAWEI",
        "info": "wifi",
        "index": "4",
        "rainbow": "",
        "schemaType": "auction",
        "elderHome": "false",
        "isEnterSrpSearch": "true",
        "newSearch": "false",
        "network": "wifi",
        "subtype": "",
        "hasPreposeFilter": "false",
        "prepositionVersion": "v2",
        "client_os": "Android",
        "gpsEnabled": "false",
        "searchDoorFrom": "srp",
        "debug_rerankNewOpenCard": "false",
        "homePageVersion": "v7",
        "searchElderHomeOpen": "false",
        "search_action": "initiative",
        "sugg": "_4_1",
        "sversion": "13.6",
        "style": "list",
        "ttid": "600000@taobao_pc_10.7.0",
        "needTabs": "true",
        "areaCode": "CN",
        "vm": "nw",
        "countryNum": "156",
        "m": "pc",
        "page": page,
        "n": 48,
        "q": keyword,
        "qSource": "url",
        "pageSource": "",
        "tab": "all",
        "pageSize": "48",
        "totalPage": "100",
        "totalResults": "5000",
        "sourceS": "48",
        "sort": "_coefp",
        "bcoffset": "-26",
        "ntoffset": "0",
        "filterTag": "",
        "service": "",
        "prop": "",
        "loc": "",
        "start_price": None,
        "end_price": None,
        "startPrice": None,
        "endPrice": None,
        "categoryp": "",
        "ha3Kvpairs": None,
        "couponFilter": 0,
        "myCNA": "4PjnHzPgIA0CARsm5jekDfQ+"
    }
    # 移除值为 None 的键
    params = {k: v for k, v in params.items() if v is not None}
    return params


def fetch_page(session: requests.Session, keyword: str, page: int, em_token: str) -> Optional[List[Dict]]:
    """请求一页商品数据，返回 itemsArray 列表，失败返回 None"""
    timestamp = str(int(time.time() * 1000))
    app_key = "12574478"

    # 构造 data 结构
    search_params = build_search_params(keyword, page)
    params_json = json.dumps(search_params)
    data_wrapper = {
        "appId": "34385",
        "params": params_json
    }

    # 生成签名
    sign, data_str = build_signature(em_token, timestamp, app_key, data_wrapper)

    # 请求参数
    query_params = {
        'jsv': '2.7.4',
        'appKey': app_key,
        't': timestamp,
        'sign': sign,
        'api': 'mtop.relationrecommend.wirelessrecommend.recommend',
        'v': '2.0',
        'timeout': '10000',
        'type': 'jsonp',
        'dataType': 'jsonp',
        'callback': 'mtopjsonp6',
        'data': data_str
    }

    url = "https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
        "Referer": "https://s.taobao.com/",
    }

    try:
        resp = session.get(url, params=query_params, headers=headers, timeout=10)
        resp.raise_for_status()
    except Exception as e:
        print(f"请求失败 (关键词={keyword}, 页码={page}): {e}")
        return None

    # 解析 JSONP 响应
    text = resp.text
    jsonp_match = re.search(r'mtopjsonp\d+\((.*)\)', text)
    if not jsonp_match:
        print(f"响应格式异常 (关键词={keyword}, 页码={page}): {text[:200]}")
        return None

    try:
        data = json.loads(jsonp_match.group(1))
    except json.JSONDecodeError:
        print(f"JSON 解析失败: {text[:200]}")
        return None

    # 检查返回状态
    ret = data.get('ret', [])
    if not ret or ret[0] != 'SUCCESS::调用成功':
        print(f"接口返回错误 (关键词={keyword}, 页码={page}): {ret}")
        return None

    items = data.get('data', {}).get('itemsArray', [])
    if not items:
        print(f"未获取到商品数据 (关键词={keyword}, 页码={page})")
    return items


def save_to_csv(items: List[Dict], filename: str):
    """保存商品数据到 CSV"""
    if not items:
        print("无数据可保存")
        return
    fieldnames = ['title', 'img', 'price', 'procity', 'realSales', 'shopName']
    with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for item in items:
            row = {
                'title': item.get('title', '').replace('<span class=H>', '').replace('</span>', ''),
                'img': item.get('pic_path', ''),
                'price': item.get('price', ''),
                'procity': item.get('procity', ''),
                'realSales': item.get('realSales', ''),
                'shopName': item.get('nick', ''),
            }
            writer.writerow(row)
    print(f"已保存 {len(items)} 条数据到 {filename}")


def main():
    # 1. 加载 cookies.txt
    try:
        session = load_cookies_from_netscape(COOKIE_FILE)
    except RuntimeError as e:
        print(e)
        return

    # 2. 提取 _m_h5_tk token
    try:
        em_token = extract_token_from_session(session)
        print(f"提取到的 token: {em_token}")
    except ValueError as e:
        print(e)
        return

    # 3. 分页爬取
    all_items = []
    for page in range(1, MAX_PAGE + 1):
        print(f"正在爬取第 {page} 页，关键词：{KEYWORD}")
        items = fetch_page(session, KEYWORD, page, em_token)
        if items is None:
            print(f"第 {page} 页爬取失败，停止后续翻页")
            break
        if not items:
            print(f"第 {page} 页无数据，停止翻页")
            break
        all_items.extend(items)
        print(f"第 {page} 页获取到 {len(items)} 个商品")
        time.sleep(REQUEST_INTERVAL)

    # 4. 保存结果
    if all_items:
        save_to_csv(all_items, OUTPUT_CSV)
    else:
        print("未获取到任何商品数据")


if __name__ == "__main__":
    main()