本教程基于 Python 脚本，可批量获取淘宝生意参谋（sycm.taobao.com）的市场排行数据（成交金额榜、增长榜、流量榜等），并将 API 返回的嵌套 JSON 字段自动展平为 CSV 表格列，同时根据排名变化字段生成“升X名/降X名/持平”的直观列。
requests 库：pip install requestsbx-ua 参数https://sycm.taobao.com，确保能正常访问市场排行页面。rank.json 或 search.json 请求。key1=value1; key2=value2）。cookie.txt，与脚本放在同一目录下。Cookie: 前缀，不要有多余换行。rank.json 请求，复制 bx-ua 请求头的值（很长的字符串）。--bx-ua 参数传入。token：通常固定为 de1acdc6b，如有变化请从抓包中更新。cateId：你有权访问的类目 ID（默认示例 50011158）。dateRange：日期范围，格式 YYYY-MM-DD|YYYY-MM-DD。将以下代码保存为 sycm_cli.py，与 cookie.txt 放在同一目录。
python#!/usr/bin/env python3
"""
淘宝生意参谋数据采集命令行工具
用法示例:
python sycm_cli.py --rank-type gmv --page 1 --output data.csv
python sycm_cli.py --rank-type growth --date-range 2026-04-13|2026-04-19 --cate-id 50011158 --max-pages 5 --output all_pages.csv
"""
import argparse
import csv
import json
import time
import sys
import os
from typing import Dict, Any, List, Optional
import requests
from requests.exceptions import RequestException
# ======================== 固定配置（可根据需要修改） ========================
# 这些是反爬字段，可能需要定期更新。也可以从命令行参数或文件读取，这里先硬编码示例。
DEFAULT_BX_V = "2.5.36"
DEFAULT_BX_UA = "你的真实bx-ua值"
DEFAULT_ONETRACE_CARD_ID = "%2Fmc%2Ffree%2Fmarket_rank%7C%E5%B8%82%E5%9C%BA%E6%8E%92%E8%A1%8C-%E5%95%86%E5%93%81-%E5%95%86%E5%93%81%E6%8E%92%E8%A1%8C"
DEFAULT_SYCM_REFERER = "/mc/free/market_rank"
DEFAULT_TOKEN = "de1acdc6b" # 这个token可能固定
DEFAULT_REFERER_URL = "https://sycm.taobao.com/mc/free/market_rank?activeKey=item&dateRange={dateRange}&dateType=recent7&parentCateId=2132&cateId={cateId}"
DEFAULT_COOKIE_FILE = "cookie.txt" # 默认 Cookie 文件路径
# ======================== 读取 Cookie ========================
def load_cookie_from_file(filename: str = "cookie.txt") -> str:
"""
从脚本所在目录读取 Cookie 文件，返回干净的 Cookie 字符串。
自动去除可能存在的 'Cookie:' 前缀、换行符、首尾空格。
"""
# 获取脚本所在目录（兼容打包后的情况）
if getattr(sys, 'frozen', False):
script_dir = os.path.dirname(sys.executable)
else:
script_dir = os.path.dirname(os.path.abspath(__file__))
filepath = os.path.join(script_dir, filename)
if not os.path.exists(filepath):
raise FileNotFoundError(f"Cookie 文件不存在: {filepath}")
with open(filepath, 'r', encoding='utf-8') as f:
raw = f.read()
# 去除可能的多行内容，只取第一行（或者全部拼接，这里简单取全部并去除换行）
cookie_str = raw.strip()
if not cookie_str:
raise ValueError("Cookie 文件为空")
# 去除可能的前缀 "Cookie:"（不区分大小写，允许后面有空格）
if cookie_str.lower().startswith('cookie:'):
cookie_str = cookie_str[7:].lstrip()
# 最终确保字符串末尾没有多余的分号或空格
cookie_str = cookie_str.rstrip('; ')
return cookie_str
# ======================== 核心请求函数 ========================
def fetch_rank_data(
rank_type: str,
page: int = 1,
cookie_str: str = "",
bx_ua: str = DEFAULT_BX_UA,
date_range: str = "2026-04-13|2026-04-19",
date_type: str = "recent7",
cate_id: str = "50011158",
page_size: int = 20,
index_code: str = "payByrCnt,uv",
token: str = DEFAULT_TOKEN,
**kwargs
) -> Dict[str, Any]:
"""
发送请求获取市场排行数据
"""
# 根据 rank_type 选择 URL
url_map = {
'gmv': "https://sycm.taobao.com/mc/mq/mkt/item/offline/rank.json",
'growth': "https://sycm.taobao.com/mc/mq/mkt/item/offline/rank.json",
'newitm_ipv': "https://sycm.taobao.com/mc/mq/mkt/item/offline/rank.json",
'flow': "https://sycm.taobao.com/mc/mq/mkt/item/offline/rank/search.json",
'add': "https://sycm.taobao.com/mc/mq/mkt/item/offline/rank/purpose.json",
}
if rank_type not in url_map:
raise ValueError(f"不支持的 rank_type: {rank_type}，可选 {list(url_map.keys())}")
url = url_map[rank_type]
# 构建 referer
referer = DEFAULT_REFERER_URL.format(dateRange=date_range, cateId=cate_id)
# 请求头
headers = {
"Host": "sycm.taobao.com",
"sec-ch-ua-platform": "\"Windows\"",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/147.0.0.0 Safari/537.36",
"accept": "*/*",
"referer": referer,
"accept-language": "zh-CN,zh;q=0.9",
"bx-v": DEFAULT_BX_V,
"bx-ua": bx_ua,
"onetrace-card-id": DEFAULT_ONETRACE_CARD_ID,
"sycm-referer": DEFAULT_SYCM_REFERER,
}
# 查询参数
params = {
"dateRange": date_range,
"dateType": date_type,
"pageSize": page_size,
"page": page,
"cateId": cate_id,
"rankType": rank_type,
"minPrice": kwargs.get('minPrice', ""),
"maxPrice": kwargs.get('maxPrice', ""),
"priceSeg": kwargs.get('priceSeg', ""),
"sellerType": kwargs.get('sellerType', "-1"),
"keyWord": kwargs.get('keyWord', ""),
"cateFlag": kwargs.get('cateFlag', "0"),
"indexCode": index_code,
"marketVersion": kwargs.get('marketVersion', "free"),
"token": token,
"_": int(time.time() * 1000),
}
# 使用 Session，并设置 Cookie
session = requests.Session()
# 将 Cookie 字符串解析为字典并更新 session.cookies
# 注意：Cookie 字符串格式如 "key1=value1; key2=value2"
for cookie_pair in cookie_str.split(';'):
cookie_pair = cookie_pair.strip()
if '=' in cookie_pair:
key, value = cookie_pair.split('=', 1)
session.cookies.set(key, value)
# 可选：忽略 SSL 警告（如果通过代理或测试）
# session.verify = False
# 如果你想消除警告，可以添加：
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
session.verify = False
# 发送请求
response = session.get(url, headers=headers, params=params)
response.raise_for_status()
# 发送请求
response = session.get(url, headers=headers, params=params)
return response.json()
# ======================== 数据提取 ========================
def extract_items(response_json: Dict[str, Any]) -> List[Dict[str, Any]]:
"""从 API 响应中提取商品列表"""
code = response_json.get('code')
if code != 0 and code != '0':
raise Exception(f"业务错误: {response_json.get('message', '未知错误')}")
data = response_json.get('data', {})
items = data.get('list')
if items is None:
items = data.get('data', [])
return items
def parse_rank_change(cycle_cqc):
"""将 cycleCqc 数值转换为中文描述"""
if cycle_cqc is None:
return ''
try:
val = int(cycle_cqc)
if val > 0:
return f'升{val}名'
elif val < 0:
return f'降{-val}名'
else:
return '持平'
except:
return ''
def flatten_item(item, parent_key='', sep='.'):
"""
递归展平字典，将嵌套的字典拆分为独立列。
例如：{"cateRankId": {"cycleCqc": 0, "value": 1}} 变为
{"cateRankId.cycleCqc": 0, "cateRankId.value": 1}
同时处理列表：转为 JSON 字符串。
"""
flattened = {}
for k, v in item.items():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
if isinstance(v, dict):
# 递归展平字典
flattened.update(flatten_item(v, new_key, sep))
elif isinstance(v, list):
# 列表转为 JSON 字符串，保持可读性
flattened[new_key] = json.dumps(v, ensure_ascii=False)
else:
flattened[new_key] = v
return flattened
def save_to_csv(items: List[Dict[str, Any]], filepath: str):
"""保存为 CSV，展平所有嵌套字段，并添加排名变化列"""
if not items:
print("没有数据可保存")
return
# 先展平每个 item
flattened_items = []
for item in items:
flat = flatten_item(item)
# 如果存在 cateRankId.cycleCqc 字段，提取并添加排名变化列
cycle_cqc = flat.get('cateRankId.cycleCqc')
if cycle_cqc is not None:
flat['排名变化'] = parse_rank_change(cycle_cqc)
# 可选：也可以将原始的 cateRankId.cycleCqc 保留（但通常不需要）
# 如果想删除原 cycleCqc 列，可以 del flat['cateRankId.cycleCqc']
flattened_items.append(flat)
# 收集所有列名
fieldnames = set()
for flat in flattened_items:
fieldnames.update(flat.keys())
# 指定希望的列顺序（可根据需要调整）
desired_order = ['排名变化', 'cateRankId.value', 'coreKeyWord.value', 'isMonitor.value',
'isSelfItem.value', 'item.title', 'item.itemId', 'payByrCnt.value',
'shop.title', 'uv.value']
# 实际存在的列 + 按 desired_order 排序，其余保持字母顺序
ordered_fieldnames = [col for col in desired_order if col in fieldnames]
remaining = sorted([col for col in fieldnames if col not in ordered_fieldnames])
fieldnames = ordered_fieldnames + remaining
with open(filepath, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore')
writer.writeheader()
for flat in flattened_items:
# 确保所有值都是字符串（CSV 写入要求）
row = {k: (v if v is not None else '') for k, v in flat.items()}
writer.writerow(row)
print(f"✅ 已保存 {len(flattened_items)} 条数据到 {filepath}")
def save_to_json(items: List[Dict[str, Any]], filepath: str):
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(items, f, ensure_ascii=False, indent=2)
print(f"✅ 已保存 {len(items)} 条数据到 {filepath}")
# ======================== 命令行参数解析 ========================
def parse_args():
parser = argparse.ArgumentParser(description="淘宝生意参谋数据采集工具")
parser.add_argument("--cookie-file", default="cookie.txt", help="Cookie 文件路径，文件内容为 Cookie 字符串 (默认 cookie.txt)")
parser.add_argument("--rank-type", required=True, choices=['gmv', 'growth', 'flow', 'newitm_ipv', 'add'], help="榜单类型")
parser.add_argument("--page", type=int, default=1, help="单页页码 (当 --max-pages 未指定时使用)")
parser.add_argument("--max-pages", type=int, default=1, help="最大翻页数，大于1时自动获取多页 (默认1，即只获取指定页)")
parser.add_argument("--output", required=True, help="输出文件路径 (支持 .csv 或 .json 后缀)")
parser.add_argument("--date-range", default="2026-04-13|2026-04-19", help="日期范围，格式 YYYY-MM-DD|YYYY-MM-DD")
parser.add_argument("--date-type", default="recent7", help="日期类型，如 recent7")
parser.add_argument("--cate-id", default="50011158", help="类目ID")
parser.add_argument("--page-size", type=int, default=20, help="每页数量 (默认20)")
parser.add_argument("--index-code", default="payByrCnt,uv", help="指标代码，逗号分隔")
parser.add_argument("--token", default="de1acdc6b", help="token参数")
parser.add_argument("--bx-ua", default=DEFAULT_BX_UA, help="bx-ua 指纹值 (若不提供则使用默认占位符，可能无效)")
parser.add_argument("--delay", type=float, default=0.5, help="翻页间隔秒数")
return parser.parse_args()
# ======================== 主函数 ========================
def main():
args = parse_args()
# 1. 读取 Cookie
try:
cookie_str = load_cookie_from_file(args.cookie_file)
print(f"✅ 已从 {args.cookie_file} 读取 Cookie (长度 {len(cookie_str)})")
except Exception as e:
print(f"❌ 读取 Cookie 失败: {e}")
sys.exit(1)
# 2. 根据 max-pages 决定获取单页还是多页
all_items = []
if args.max_pages == 1:
# 单页模式
print(f"正在获取 {args.rank_type} 榜第 {args.page} 页...")
try:
resp = fetch_rank_data(
rank_type=args.rank_type,
page=args.page,
cookie_str=cookie_str,
bx_ua=args.bx_ua,
date_range=args.date_range,
date_type=args.date_type,
cate_id=args.cate_id,
page_size=args.page_size,
index_code=args.index_code,
token=args.token,
)
items = extract_items(resp)
print(f"✅ 获取成功，共 {len(items)} 条数据")
all_items = items
except RequestException as e:
print(f"❌ 网络请求失败: {e}")
sys.exit(1)
except Exception as e:
print(f"❌ 错误: {e}")
sys.exit(1)
else:
# 多页自动翻页模式
print(f"开始自动翻页获取 {args.rank_type} 榜，最大页数 {args.max_pages}...")
for page in range(1, args.max_pages + 1):
print(f" 正在获取第 {page} 页...")
try:
resp = fetch_rank_data(
rank_type=args.rank_type,
page=page,
cookie_str=cookie_str,
bx_ua=args.bx_ua,
date_range=args.date_range,
date_type=args.date_type,
cate_id=args.cate_id,
page_size=args.page_size,
index_code=args.index_code,
token=args.token,
)
items = extract_items(resp)
if not items:
print(f" 第 {page} 页无数据，停止翻页")
break
all_items.extend(items)
print(f" 本页 {len(items)} 条，累计 {len(all_items)} 条")
if len(items) < args.page_size:
print(" 已到达最后一页")
break
time.sleep(args.delay)
except Exception as e:
print(f" 第 {page} 页请求失败: {e}")
break
# 3. 保存数据
if not all_items:
print("没有获取到任何数据，退出")
sys.exit(0)
# 根据输出文件扩展名选择保存格式
output_lower = args.output.lower()
if output_lower.endswith('.csv'):
save_to_csv(all_items, args.output)
elif output_lower.endswith('.json'):
save_to_json(all_items, args.output)
else:
# 默认保存为 CSV
save_to_csv(all_items, args.output + '.csv')
print(f"⚠️ 未识别扩展名，已保存为 {args.output}.csv")
if __name__ == '__main__':
main()
将浏览器中复制的 Cookie 字符串（不含 Cookie: 前缀）保存为 cookie.txt，放在脚本同级目录。
DEFAULT_BX_UA 替换为你抓包得到的真实 bx-ua 值。token 有变化，一并修改 DEFAULT_TOKEN。基本语法：
bashpython sycm_cli.py --rank-type <榜单类型> [选项] --output <输出文件>
常用示例：
| 目的 | 命令 |
|---|---|
| 获取成交金额榜第1页，输出 CSV | python sycm_cli.py --rank-type gmv --page 1 --output result.csv |
| 获取增长榜前3页，输出 CSV | python sycm_cli.py --rank-type growth --max-pages 3 --output growth.csv |
| 获取流量榜第2页，每页30条 | python sycm_cli.py --rank-type flow --page 2 --page-size 30 --output flow.csv |
| 自定义日期和类目 | python sycm_cli.py --rank-type gmv --date-range "2026-04-01|2026-04-07" --cate-id 12345678 --output custom.csv |
| 输出 JSON 格式（原始未展平） | python sycm_cli.py --rank-type gmv --page 1 --output data.json |
| 参数 | 说明 | 默认值 |
|---|---|---|
--rank-type | 榜单类型：gmv, growth, flow, newitm_ipv, add | 必填 |
--page | 页码 | 1 |
--max-pages | 最大翻页数（>1 时自动多页） | 1 |
--output | 输出文件路径（.csv 或 .json） | 必填 |
--cookie-file | Cookie 文件路径 | cookie.txt |
--date-range | 日期范围 开始|结束 | 示例值 |
--date-type | 日期类型（如 recent7） | recent7 |
--cate-id | 类目 ID | 50011158 |
--page-size | 每页数量 | 20 |
--index-code | 指标代码（逗号分隔） | payByrCnt,uv |
--token | token 参数 | de1acdc6b |
--bx-ua | bx-ua 指纹值（必须有效） | 默认占位符 |
--delay | 翻页间隔（秒） | 0.5 |
. 连接（例如 cateRankId.value、item.title）。cateRankId.cycleCqc 生成“升X名”、“降X名”或“持平”。desired_order 变量）。示例 CSV 列（部分）：
排名变化, cateRankId.value, coreKeyWord.value, item.title, payByrCnt.value, uv.value, ... 升216名, 51, 双人 泡澡桶 木桶, 双人泡澡桶木桶..., 3, 25~50, ...
bx-ua 无效或已过期，Cookie 失效。bx-ua 和 Cookie，通过 --bx-ua 传入或修改代码中的 DEFAULT_BX_UA，并更新 cookie.txt。login.taobao.com）cookie.txt。cateRankId.cycleCqc 字段，或字段名不一致。parse_rank_change 中的字段路径。--delay，避免高频请求导致 IP 被封。bx-ua 均有有效期，建议每次运行前更新。通过以上步骤，你可以轻松获取生意参谋的市场排行数据，并获得清晰展平的 CSV 表格。如有其他问题，欢迎在技术社区交流。
本文作者:苏皓明
本文链接:
版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！