西域平台关键字搜索接口实战：工业用品数据获取与解析方案

在工业采购与供应链管理领域，西域西域（Xiyuwang）作为专业的 MRO 工业用品采购平台，其商品数据对于工业电商系统、采购分析工具等应用具有重要价值。本文将详细介绍西域平台关键字搜索接口的调用方法，涵盖认证机制、参数配置、分页处理及数据解析的完整流程，并提供可直接复用的 Python 代码实现，帮助开发者快速对接西域开放平台。
一、西域搜索接口基础信息

西域提供的商品搜索接口（/api/v1/products/search）是获取工业用品列表的核心接口，支持多维度筛选条件，特别适用于工业零部件、工具设备等商品的精准检索。

接口特点：

    采用 API Key+Token 的双重认证机制
    支持按关键字、分类、品牌、价格、库存等多条件组合筛选
    分页加载数据，最大页容量为 50 条
    响应包含商品基本信息、规格参数、价格、库存、供应商等工业用品特有数据

接口端点：https://api.xiyuwang.com/api/v1/products/search
二、认证机制与参数说明
1. 认证流程

西域接口采用基于 API Key 和 Token 的认证方式，具体流程如下：

    在西域开发者平台注册应用，获取 API Key
    通过 API Key 调用令牌接口获取访问 Token（有效期 24 小时）
    在所有请求的 Header 中携带 Token 进行身份验证

2. 核心参数说明

    公共 Header 参数：
        X-API-Key：应用的 API Key
        Authorization：令牌，格式为Bearer {token}

    查询参数：
        keyword：搜索关键字（必填）
        page：页码（默认 1）
        size：每页条数（1-50，默认 20）
        categoryId：分类 ID（可选）
        brandId：品牌 ID（可选）
        priceMin/priceMax：价格区间（可选）
        stockMin：最低库存（可选）
        sort：排序方式（default - 默认，priceAsc - 价升，priceDesc - 价降，stockDesc - 库存降）
        spec：规格筛选（可选，如 "voltage:220V"）

点击获取key和secret

三、完整代码实现

以下是 Python 实现的西域平台关键字搜索功能，包含令牌管理、多条件筛选和结果解析：

    import requests
    import time
    from typing import Dict, List, Optional, Any
     
    class XiyuwangSearchAPI:
        def __init__(self, api_key: str, api_secret: str):
            """
            初始化西域搜索API客户端
            
            :param api_key: 应用的API Key
            :param api_secret: 应用的API Secret
            """
            self.api_key = api_key
            self.api_secret = api_secret
            self.base_url = "https://api.xiyuwang.com"
            self.search_endpoint = "/api/v1/products/search"
            self.token_endpoint = "/api/v1/auth/token"
            self.max_page_size = 50  # 最大每页条数
            self.token = None
            self.token_expiry = 0  # 令牌过期时间戳(秒)
        
        def _get_token(self) -> Optional[str]:
            """获取或刷新访问令牌"""
            # 检查令牌是否有效（提前300秒刷新）
            if self.token and time.time() < self.token_expiry - 300:
                return self.token
                
            # 准备请求参数
            headers = {
                "X-API-Key": self.api_key,
                "Content-Type": "application/json"
            }
            
            payload = {
                "secret": self.api_secret,
                "grantType": "client_credentials"
            }
            
            try:
                response = requests.post(
                    f"{self.base_url}{self.token_endpoint}",
                    json=payload,
                    headers=headers,
                    timeout=10
                )
                response.raise_for_status()
                result = response.json()
                
                if result.get("success"):
                    self.token = result.get("data", {}).get("token")
                    expires_in = result.get("data", {}).get("expiresIn", 86400)  # 默认24小时
                    self.token_expiry = time.time() + expires_in
                    return self.token
                else:
                    print(f"获取令牌失败: {result.get('message')}")
                    return None
                    
            except requests.exceptions.RequestException as e:
                print(f"令牌请求异常: {str(e)}")
                return None
        
        def search_products(self,
                           keyword: str,
                           page: int = 1,
                           size: int = 20,
                           category_id: Optional[str] = None,
                           brand_id: Optional[str] = None,
                           price_min: Optional[float] = None,
                           price_max: Optional[float] = None,
                           stock_min: Optional[int] = None,
                           sort: str = "default",
                           spec: Optional[str] = None) -> Dict[str, Any]:
            """
            搜索西域平台商品
            
            :param keyword: 搜索关键字
            :param page: 页码
            :param size: 每页条数(1-50)
            :param category_id: 分类ID
            :param brand_id: 品牌ID
            :param price_min: 最低价格
            :param price_max: 最高价格
            :param stock_min: 最低库存
            :param sort: 排序方式
            :param spec: 规格筛选
            :return: 搜索结果
            """
            # 获取令牌
            token = self._get_token()
            if not token:
                return {
                    "success": False,
                    "error_msg": "无法获取访问令牌"
                }
            
            # 限制最大页容量
            size = min(size, self.max_page_size)
            
            # 构建查询参数
            params: Dict[str, Any] = {
                "keyword": keyword,
                "page": page,
                "size": size,
                "sort": sort
            }
            
            # 添加可选参数
            if category_id:
                params["categoryId"] = category_id
            if brand_id:
                params["brandId"] = brand_id
            if price_min is not None:
                params["priceMin"] = price_min
            if price_max is not None:
                params["priceMax"] = price_max
            if stock_min is not None:
                params["stockMin"] = stock_min
            if spec:
                params["spec"] = spec
            
            # 设置请求头
            headers = {
                "X-API-Key": self.api_key,
                "Authorization": f"Bearer {token}",
                "Accept": "application/json"
            }
            
            try:
                # 发送请求
                response = requests.get(
                    f"{self.base_url}{self.search_endpoint}",
                    params=params,
                    headers=headers,
                    timeout=15
                )
                response.raise_for_status()
                
                # 解析响应
                result = response.json()
                
                # 处理API错误
                if not result.get("success"):
                    return {
                        "success": False,
                        "error_code": result.get("code"),
                        "error_msg": result.get("message", "未知错误")
                    }
                    
                # 处理正常响应
                return self._parse_search_result(result.get("data", {}))
                
            except requests.exceptions.RequestException as e:
                return {
                    "success": False,
                    "error_msg": f"请求异常: {str(e)}"
                }
            except Exception as e:
                return {
                    "success": False,
                    "error_msg": f"处理响应失败: {str(e)}"
                }
        
        def _parse_search_result(self, raw_data: Dict[str, Any]) -> Dict[str, Any]:
            """
            解析原始搜索结果为结构化数据
            
            :param raw_data: 原始API响应数据
            :return: 结构化的搜索结果
            """
            # 提取分页信息
            pagination = {
                "total": raw_data.get("total", 0),
                "page": raw_data.get("page", 1),
                "size": raw_data.get("size", 20),
                "pages": raw_data.get("pages", 0)
            }
            
            # 解析商品列表
            products = []
            for item in raw_data.get("list", []):
                # 解析规格参数
                specifications = []
                for spec in item.get("specifications", []):
                    specifications.append({
                        "name": spec.get("name"),
                        "value": spec.get("value")
                    })
                
                # 解析价格信息
                price_info = {
                    "price": item.get("price"),
                    "original_price": item.get("originalPrice"),
                    "currency": item.get("currency", "CNY"),
                    "moq": item.get("moq", 1),  # 最小起订量
                    "discount": item.get("discount")  # 折扣信息
                }
                
                products.append({
                    "product_id": item.get("id"),
                    "sku": item.get("sku"),
                    "name": item.get("name"),
                    "model": item.get("model"),  # 型号
                    "image": item.get("mainImage"),
                    "brand": {
                        "id": item.get("brandId"),
                        "name": item.get("brandName")
                    },
                    "category": {
                        "id": item.get("categoryId"),
                        "name": item.get("categoryName")
                    },
                    "price": price_info,
                    "stock": {
                        "quantity": item.get("stockQuantity"),
                        "warehouse": item.get("warehouseName")
                    },
                    "specifications": specifications,
                    "certifications": item.get("certifications", []),  # 认证信息
                    "delivery": {
                        "lead_time": item.get("leadTime"),  # 交货期
                        "support_batch": item.get("supportBatch", False)  # 是否支持批量
                    },
                    "url": item.get("productUrl")
                })
            
            return {
                "success": True,
                "pagination": pagination,
                "products": products,
                "filters": self._parse_filters(raw_data.get("filters", {}))
            }
        
        def _parse_filters(self, raw_filters: Dict[str, Any]) -> Dict[str, Any]:
            """解析可用筛选条件"""
            filters = {}
            
            # 分类筛选
            if "categories" in raw_filters:
                filters["categories"] = [
                    {"id": item.get("id"), "name": item.get("name"), "count": item.get("count")}
                    for item in raw_filters["categories"]
                ]
                
            # 品牌筛选
            if "brands" in raw_filters:
                filters["brands"] = [
                    {"id": item.get("id"), "name": item.get("name"), "count": item.get("count")}
                    for item in raw_filters["brands"]
                ]
                
            # 规格筛选
            if "specs" in raw_filters:
                specs = {}
                for spec in raw_filters["specs"]:
                    specs[spec.get("name")] = [
                        {"value": item.get("value"), "count": item.get("count")}
                        for item in spec.get("values", [])
                    ]
                filters["specs"] = specs
                
            return filters
        
        def batch_fetch_products(self,
                               keyword: str,
                               max_pages: int = 5,
                               **kwargs) -> Dict[str, Any]:
            """
            批量获取多页搜索结果
            
            :param keyword: 搜索关键字
            :param max_pages: 最大获取页数
            :param**kwargs: 其他搜索参数
            :return: 合并的搜索结果
            """
            all_products = []
            current_page = 1
            total_pages = 1
            
            while current_page <= max_pages and current_page <= total_pages:
                # 搜索当前页
                result = self.search_products(
                    keyword=keyword,
                    page=current_page,
                    **kwargs
                )
                
                if not result.get("success"):
                    return result
                    
                # 收集商品
                all_products.extend(result.get("products", []))
                
                # 更新分页信息
                pagination = result.get("pagination", {})
                total_pages = pagination.get("pages", 1)
                
                # 准备下一页
                current_page += 1
                
                # 添加请求间隔，避免触发频率限制
                time.sleep(2)
            
            return {
                "success": True,
                "total_products": len(all_products),
                "products": all_products,
                "summary": {
                    "total_pages": total_pages,
                    "fetched_pages": current_page - 1,
                    "total_available": pagination.get("total", 0)
                }
            }
     
    # 使用示例
    if __name__ == "__main__":
        # 替换为你的应用凭证
        API_KEY = "your_api_key"
        API_SECRET = "your_api_secret"
        
        # 初始化API客户端
        xiyu_api = XiyuwangSearchAPI(API_KEY, API_SECRET)
        
        # 示例1：基础搜索（工业电机）
        basic_result = xiyu_api.search_products(
            keyword="三相异步电机",
            page=1,
            size=20,
            sort="priceAsc",
            price_min=1000,
            stock_min=10
        )
        
        if basic_result["success"]:
            print(f"基础搜索: 找到 {basic_result['pagination']['total']} 个商品")
            print(f"第一页商品数量: {len(basic_result['products'])}")
            if basic_result["products"]:
                product = basic_result["products"][0]
                print(f"第一个商品: {product['name']}, 价格: {product['price']['price']}{product['price']['currency']}")
                print(f"规格: {[f'{s["name"]}:{s["value"]}' for s in product['specifications'][:3]]}")
        
        # 示例2：批量获取多页结果
        multi_page_result = xiyu_api.batch_fetch_products(
            keyword="精密轴承",
            max_pages=3,
            size=50,
            sort="stockDesc"
        )
        
        if multi_page_result["success"]:
            print(f"\n多页搜索: 共获取 {multi_page_result['total_products']} 个商品")

四、代码核心功能解析

    令牌管理机制：
        自动处理令牌的获取与过期刷新，无需手动干预
        提前 300 秒刷新令牌，避免请求时令牌过期
        封装令牌请求逻辑，简化核心搜索功能实现

    工业用品数据处理：
        专门针对工业商品特性设计数据结构，包含型号、规格、认证等特有字段
        解析最小起订量（MOQ）、交货期等工业采购关键信息
        提取产品认证信息，满足工业采购合规性要求

    搜索功能增强：
        支持规格参数筛选，满足工业用品精准搜索需求
        提供批量获取多页数据的高级方法
        内置请求间隔控制，符合平台调用规范

    结构化解析：
        将原始响应数据转换为清晰的层级结构
        分类组织商品信息（基础信息、价格、库存、规格等）
        解析可用筛选条件，便于前端实现高级筛选功能

五、实战注意事项

    接口权限与申请：
        西域 API 需要企业资质申请，主要面向工业供应链相关企业
        不同类型的应用有不同的调用配额，工业企业应用配额更高
        部分特殊品类（如特种设备）的数据需要额外资质审核

    搜索策略优化：
        工业用品搜索建议结合分类 ID 和规格参数，提高精准度
        批量获取数据时，注意控制请求频率（建议间隔 2 秒以上）
        对常用物料建立本地缓存，缓存时间建议设置为 24 小时

    数据处理建议：
        型号和规格是工业用品的核心标识，需重点处理和存储
        注意区分不同包装规格的价格差异
        库存数据具有时效性，关键采购场景建议实时获取

    合规与安全：
        严格遵守西域平台的数据使用协议，不得用于非工业采购场景
        API Key 和 Secret 需妥善保管，避免泄露
        生产环境建议部署在服务端，避免客户端直接调用

六、功能扩展方向

    集成商品详情接口，获取完整的技术参数和供应商信息
    开发价格对比功能，分析不同规格、品牌的性价比
    实现库存预警功能，监控关键物料的库存水平
    开发采购推荐系统，基于历史搜索和规格参数推荐合适产品

通过本文提供的方案，开发者可以快速实现西域平台的工业用品搜索功能，为工业电商、采购管理系统等应用提供精准的数据支持。实际开发中，建议根据工业采购的业务特性，优化搜索参数和数据处理逻辑，提升采购效率。