微店的item_search接口是用于根据关键词搜索商品列表的核心接口，能够根据用户输入的关键词，返回相关的商品列表信息。该接口对于电商平台的商品检索索功能、竞品竞品分析、市场调研调研等场景具有重要意义。
接口功能与定位
认证机制
核心参数与响应结构
请求参数
响应核心字段
Python 脚本实现
import requests
import time
import json
import logging
from typing import Dict, Optional, List
from requests.exceptions import RequestException
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
class WeidianItemSearchAPI:
def __init__(self, appkey: str, appsecret: str):
"""
初始化微店商品搜索API客户端
:param appkey: 微店开放平台appkey
:param appsecret: 微店开放平台appsecret
"""
self.appkey = appkey
self.appsecret = appsecret
self.base_url = "https://api.vdian.com"
self.access_token = None
self.token_expires_at = 0 # token过期时间戳
self.session = requests.Session()
def _get_access_token(self) -> Optional[str]:
"""获取访问令牌"""
# 检查token是否有效
if self.access_token and self.token_expires_at > time.time() + 60:
return self.access_token
logging.info("获取新的access_token")
url = f"{self.base_url}/token"
params = {
"grant_type": "client_credentials",
"appkey": self.appkey,
"appsecret": self.appsecret
}
try:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
result = response.json()
if result.get("status") == 0:
self.access_token = result.get("result", {}).get("access_token")
expires_in = result.get("result", {}).get("expires_in", 3600)
self.token_expires_at = time.time() + expires_in
return self.access_token
else:
logging.error(f"获取access_token失败: {result.get('msg')}")
return None
except RequestException as e:
logging.error(f"获取access_token请求异常: {str(e)}")
return None
def search_items(self, keyword: str, page: int = 1, page_size: int = 20, sort: Optional[str] = None) -> Optional[Dict]:
"""
根据关键词搜索商品
:param keyword: 搜索关键词
:param page: 页码
:param page_size: 每页商品数量
:param sort: 排序方式
:return: 商品搜索结果
"""
# 获取有效的access_token
if not self._get_access_token():
return None
url = f"{self.base_url}/item/search"
# 构建请求参数
params = {
"keyword": keyword,
"page": page,
"page_size": page_size,
"access_token": self.access_token
}
# 添加排序参数
if sort:
params["sort"] = sort
try:
response = self.session.get(url, params=params, timeout=15)
response.raise_for_status()
result = response.json()
if result.get("status") == 0:
# 格式化商品数据
return self._format_search_result(result.get("result", {}))
else:
logging.error(f"商品搜索失败: {result.get('msg')} (错误码: {result.get('status')})")
return None
except RequestException as e:
logging.error(f"商品搜索请求异常: {str(e)}")
return None
except json.JSONDecodeError:
logging.error(f"商品搜索响应解析失败: {response.text[:200]}...")
return None
def _format_search_result(self, result_data: Dict) -> Dict:
"""格式化搜索结果数据"""
# 分页信息
pagination = {
"total_items": int(result_data.get("total", 0)),
"total_pages": (int(result_data.get("total", 0)) + int(result_data.get("page_size", 20)) - 1) // int(result_data.get("page_size", 20)),
"current_page": int(result_data.get("page", 1)),
"page_size": int(result_data.get("page_size", 20))
}
# 商品列表
items = []
for item in result_data.get("items", {}).get("item", []):
items.append({
"item_id": item.get("item_id"),
"title": item.get("title"),
"main_img": item.get("main_img"),
"price": float(item.get("price", 0)),
"original_price": float(item.get("original_price", 0)),
"sales": int(item.get("sales", 0)),
"shop_id": item.get("shop_id"),
"shop_name": item.get("shop_name"),
"detail_url": item.get("detail_url")
})
return {
"pagination": pagination,
"items": items
}
def search_all_items(self, keyword: str, max_pages: int = 5) -> List[Dict]:
"""
搜索所有页的商品
:param keyword: 搜索关键词
:param max_pages: 最大页数限制
:return: 所有商品列表
"""
all_items = []
page = 1
while page <= max_pages:
logging.info(f"搜索关键词 '{keyword}' 第 {page} 页商品")
result = self.search_items(keyword, page=page, page_size=100)
if not result or not result["items"]:
break
all_items.extend(result["items"])
# 检查是否已达最后一页
if page >= result["pagination"]["total_pages"]:
break
page += 1
# 添加延迟，避免触发频率限制
time.sleep(1)
return all_items
# 示例调用
if __name__ == "__main__":
# 替换为实际的appkey和appsecret（从微店开放平台获取）
APPKEY = "your_appkey"
APPSECRET = "your_appsecret"
# 搜索关键词
KEYWORD = "手机"
# 初始化API客户端
api = WeidianItemSearchAPI(APPKEY, APPSECRET)
# 方式1：获取指定页的商品
# search_result = api.search_items(KEYWORD, page=1, page_size=20, sort="sales_desc")
# 方式2：获取多页商品
search_result = api.search_all_items(KEYWORD, max_pages=3)
if isinstance(search_result, dict) and "items" in search_result:
print(f"关键词 '{KEYWORD}' 搜索结果：")
print(f"总商品数：{search_result['pagination']['total_items']}")
print(f"总页数：{search_result['pagination']['total_pages']}")
print(f"当前页：{search_result['pagination']['current_page']}")
print("\n前5件商品信息：")
for i, item in enumerate(search_result["items"][:5], 1):
print(f"{i}. {item['title']}")
print(f" 价格：{item['price']}元，原价：{item['original_price']}元")
print(f" 销量：{item['sales']}件")
print(f" 店铺：{item['shop_name']}")
print(f" 链接：{item['detail_url']}")
print("-" * 80)
elif isinstance(search_result, list):
print(f"关键词 '{KEYWORD}' 共获取到 {len(search_result)} 件商品")
print("\n前5件商品信息：")
for i, item in enumerate(search_result[:5], 1):
print(f"{i}. {item['title']}")
print(f" 价格：{item['price']}元，原价：{item['original_price']}元")
print(f" 销量：{item['sales']}件")
print(f" 店铺：{item['shop_name']}")
print(f" 链接：{item['detail_url']}")
print("-" * 80)