蘑菇街是国内知名的时尚电商平台，聚焦年轻女性用户，以服饰、美妆、家居等品类为核心，其商品详情数据（如售价、规格、库存、店铺信息等）是电商卖家选品、比价工具开发、市场分析的重要依据。由于蘑菇街官方 API 接入门槛较高（需企业资质且功能有限），开发者常通过页面解析或第三方服务实现商品详情（item_get）的获取。本文将系统讲解接口对接逻辑、技术实现、反爬应对及最佳实践，帮助开发者构建稳定的商品详情获取系统。
一、接口基础认知（核心功能与场景）
二、对接前置准备（环境与工具）
三、接口调用流程（基于动态解析）
四、代码实现示例（Python）
import requests
import time
import random
import re
import hashlib
import json
from fake_useragent import UserAgent
from typing import Dict, List
class MogujieItemApi:
def __init__(self, cookie: str, proxy_pool: List[str] = None):
self.base_api = "https://gateway.mogujie.com"
self.app_key = "12574478" # 固定appKey
self.cookie = cookie
self.cookie_tk = self._extract_cookie_tk() # 从Cookie中提取_m_h5_tk
self.ua = UserAgent()
self.proxy_pool = proxy_pool
def _extract_cookie_tk(self) -> str:
"""从Cookie中提取_m_h5_tk"""
for cookie in self.cookie.split(";"):
if "_m_h5_tk" in cookie:
return cookie.split("=")[1].strip()
raise ValueError("Cookie中未找到_m_h5_tk，请检查Cookie是否有效")
def _generate_sign(self, data: str) -> tuple:
"""生成加密参数_m_h5_tk和timestamp"""
token = self.cookie_tk.split("_")[0]
timestamp = str(int(time.time() * 1000))
# 拼接签名源串
sign_str = f"{token}&{timestamp}&{self.app_key}&{data}"
sign = hashlib.md5(sign_str.encode()).hexdigest()
m_h5_tk = f"{token}_{sign}"
return m_h5_tk, timestamp
def _get_headers(self) -> Dict[str, str]:
"""生成请求头"""
return {
"User-Agent": self.ua.mobile, # 优先使用移动端UA，反爬较松
"Referer": "https://m.mogujie.com/",
"Origin": "https://m.mogujie.com",
"Cookie": self.cookie,
"Content-Type": "application/json"
}
def _get_proxy(self) -> Dict[str, str]:
"""随机获取代理"""
if self.proxy_pool and len(self.proxy_pool) > 0:
proxy = random.choice(self.proxy_pool)
return {"http": proxy, "https": proxy}
return None
def _fetch_api(self, api_path: str, data: Dict) -> Dict:
"""调用蘑菇街动态API"""
try:
# 生成加密参数
data_str = json.dumps(data, ensure_ascii=False)
m_h5_tk, timestamp = self._generate_sign(data_str)
# 构建URL参数
params = {
"_m_h5_tk": m_h5_tk,
"_m_h5_tk_enc": "", # 部分接口无需enc，复杂接口需额外处理
"timestamp": timestamp,
"appKey": self.app_key
}
url = f"{self.base_api}{api_path}"
headers = self._get_headers()
proxy = self._get_proxy()
# 发送请求（带随机延迟）
time.sleep(random.uniform(1, 3))
response = requests.post(
url=url,
params=params,
data=data_str,
headers=headers,
proxies=proxy,
timeout=10
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"API请求失败: {str(e)}")
return {}
def _parse_base_info(self, base_data: Dict) -> Dict:
"""解析基础信息接口数据"""
item = base_data.get("data", {}).get("item", {})
return {
"title": item.get("title", ""),
"main_image": item.get("mainImg", ""),
"images": item.get("imgs", []),
"shop": {
"id": item.get("shopId", ""),
"name": item.get("shopName", ""),
"score": item.get("shopScore", 0)
}
}
def _parse_price_stock(self, price_data: Dict) -> Dict:
"""解析价格与库存接口数据"""
price_info = price_data.get("data", {}).get("price", {})
return {
"price": {
"original": price_info.get("originalPrice", 0),
"current": price_info.get("currentPrice", 0),
"discount": price_info.get("discount", "")
},
"total_stock": price_data.get("data", {}).get("stock", 0),
"specs": price_data.get("data", {}).get("specs", []) # 规格+库存
}
def _parse_sales(self, sales_data: Dict) -> Dict:
"""解析销量与评价接口数据"""
sales_info = sales_data.get("data", {})
return {
"monthly_sales": sales_info.get("monthlySales", 0),
"comment_count": sales_info.get("commentCount", 0),
"good_rate": sales_info.get("goodRate", 0) # 好评率（如98表示98%）
}
def item_get(self, item_id: str) -> Dict:
"""
获取蘑菇街商品详情
:param item_id: 商品ID（如12345678）
:return: 标准化商品数据
"""
try:
# 1. 获取基础信息
base_data = self._fetch_api(
api_path="/api/item/detail",
data={"itemId": item_id}
)
if not base_data.get("success"):
return {"success": False, "error_msg": "基础信息接口返回失败"}
base_info = self._parse_base_info(base_data)
# 2. 获取价格与库存
price_data = self._fetch_api(
api_path="/api/item/price",
data={"itemId": item_id}
)
price_stock = self._parse_price_stock(price_data)
# 3. 获取销量与评价
sales_data = self._fetch_api(
api_path="/api/item/rate",
data={"itemId": item_id, "page": 1, "size": 1}
)
sales_info = self._parse_sales(sales_data)
# 4. 合并数据
result = {
"success": True,
"data": {
"item_id": item_id,** base_info,
**price_stock,** sales_info,
"update_time": time.strftime("%Y-%m-%d %H:%M:%S")
}
}
return result
except Exception as e:
return {"success": False, "error_msg": f"获取失败: {str(e)}", "code": -1}
# 使用示例
if __name__ == "__main__":
# 从浏览器获取的Cookie（需包含_m_h5_tk）
COOKIE = "_m_h5_tk=xxx; _m_h5_tk_enc=xxx; ..." # 替换为实际Cookie
# 代理池（替换为有效代理）
PROXIES = [
"http://123.45.67.89:8888",
"http://98.76.54.32:8080"
]
# 初始化API客户端
api = MogujieItemApi(cookie=COOKIE, proxy_pool=PROXIES)
# 获取商品详情（示例item_id）
item_id = "12345678" # 替换为实际商品ID
result = api.item_get(item_id)
if result["success"]:
data = result["data"]
print(f"商品标题: {data['title']}")
print(f"价格: 原价{data['price']['original']}元 | 现价{data['price']['current']}元 | {data['price']['discount']}")
print(f"销量: 月销{data['monthly_sales']}件 | 评价{data['comment_count']}条 | 好评率{data['good_rate']}%")
print(f"库存: {data['total_stock']}件 | 店铺: {data['shop']['name']}（评分{data['shop']['score']}）")
print(f"主图: {data['main_image']}")
print("\n规格信息（前3条）:")
for spec in data["specs"][:3]:
print(f" {spec.get('color')} | {spec.get('size')} | 库存: {spec.get('stock')}件 | 价格: {spec.get('price')}元")
else:
print(f"获取失败: {result['error_msg']}（错误码: {result.get('code')}）")