1688 商品详情接口开发实战：从平台特性到高可用实现

import timeimport hashlibimport urllib.parsedef generate_1688_sign(params, app_secret):
    """
    生成1688接口签名
    :param params: 请求参数字典
    :param app_secret: 应用密钥
    :return: 签名字符串
    """
    # 1. 按参数名ASCII码升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    
    # 2. 拼接为key=value&key=value格式
    query_string = urllib.parse.urlencode(sorted_params)
    
    # 3. 拼接appsecret并MD5加密
    sign_str = f"{query_string}{app_secret}"
    sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    return sign

import requestsimport jsonfrom typing import Dict, Optional, Anyclass AlibabaItemAPI:
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://gw.open.1688.com/openapi/param2/2.0/"
        
    def get_item_detail(self, item_id: str, fields: list = None) -> Optional[Dict[str, Any]]:
        """
        获取1688商品详情
        :param item_id: 商品ID
        :param fields: 需要返回的字段列表，None表示返回全部
        :return: 商品详情字典
        """
        # 1. 构建基础参数
        params = {
            "method": "alibaba.item.get",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "item_id": item_id        }
        
        # 2. 添加字段筛选参数
        if fields:
            params["fields"] = ",".join(fields)
        
        # 3. 生成签名
        params["sign"] = generate_1688_sign(params, self.app_secret)
        
        try:
            # 4. 发送GET请求
            response = requests.get(
                self.api_url,
                params=params,
                timeout=15
            )
            
            # 5. 检查响应状态
            response.raise_for_status()
            
            # 6. 解析响应数据
            result = json.loads(response.text)
            
            # 7. 处理错误响应
            if "error_response" in result:
                error = result["error_response"]
                print(f"接口错误: {error.get('msg')} (错误码: {error.get('code')})")
                return None
                
            return result.get("alibaba_item_get_response", {}).get("item", {})
            
        except requests.exceptions.RequestException as e:
            print(f"HTTP请求异常: {str(e)}")
            return None
        except json.JSONDecodeError as e:
            print(f"JSON解析错误: {str(e)}")
            return None

import pandas as pdimport jsondef parse_b2b_item_detail(raw_item: Dict) -> Dict:
    """
    解析1688商品详情，提取B2B相关特性
    :param raw_item: 接口返回的原始商品数据
    :return: 结构化的商品信息字典
    """
    if not raw_item:
        return {}
    
    # 解析基础信息
    base_info = {
        "item_id": raw_item.get("item_id"),
        "title": raw_item.get("title"),
        "subject": raw_item.get("subject"),
        "price": raw_item.get("price"),
        "price_unit": raw_item.get("price_unit", "件"),
        "pic_url": raw_item.get("pic_url"),
        "detail_url": raw_item.get("detail_url"),
        "category_id": raw_item.get("category_id")
    }
    
    # 解析B2B核心信息 - 批发相关
    wholesale_info = {
        "min_buy_quantity": raw_item.get("min_buy_quantity"),  # 起订量
        "quantity": raw_item.get("quantity"),  # 总库存
        "can_mix": raw_item.get("mix_enable", False),  # 是否支持混批
        "step_price": raw_item.get("step_price", [])  # 阶梯价格
    }
    
    # 解析交易属性
    trade_props = {}
    for prop in raw_item.get("trade_props", []):
        trade_props[prop.get("name")] = prop.get("value")
    
    # 解析SKU信息
    sku_list = []
    sku_infos = raw_item.get("sku_infos", {})
    for sku in sku_infos.get("sku_info", []):
        try:
            spec = json.loads(sku.get("spec_json", "{}"))
        except json.JSONDecodeError:
            spec = {}
            
        sku_info = {
            "sku_id": sku.get("sku_id"),
            "price": sku.get("price"),
            "quantity": sku.get("quantity"),
            "spec": spec        }
        sku_list.append(sku_info)
    
    # 解析供应商信息
    seller_info = raw_item.get("seller", {})
    supplier_info = {
        "seller_id": seller_info.get("user_id"),
        "seller_nick": seller_info.get("nick"),
        "company_name": seller_info.get("company_name"),
        "credit_level": seller_info.get("credit_level"),
        "trust_year": seller_info.get("year"),  # 诚信通年限
        "is_verified": seller_info.get("is_verified", False)  # 是否实名认证
    }
    
    # 整合所有信息
    return {
        "base_info": base_info,
        "wholesale_info": wholesale_info,
        "trade_props": trade_props,
        "sku_list": sku_list,
        "supplier_info": supplier_info    }def sku_to_dataframe(sku_list: list) -> pd.DataFrame:
    """将SKU列表转换为DataFrame"""
    if not sku_list:
        return pd.DataFrame()
        
    # 展开规格字典为列
    sku_data = []
    for sku in sku_list:
        sku_dict = {
            "sku_id": sku.get("sku_id"),
            "price": sku.get("price"),
            "quantity": sku.get("quantity")
        }
        # 添加规格信息
        sku_dict.update(sku.get("spec", {}))
        sku_data.append(sku_dict)
        
    return pd.DataFrame(sku_data)

if __name__ == "__main__":
    # 配置应用信息（替换为实际值）
    APP_KEY = "your_app_key_here"
    APP_SECRET = "your_app_secret_here"
    
    # 初始化API客户端
    item_api = AlibabaItemAPI(APP_KEY, APP_SECRET)
    
    # 目标商品ID
    TARGET_ITEM_ID = "61234567890"  # 替换为实际商品ID
    
    # 指定需要返回的字段（减少数据传输量）
    REQUIRED_FIELDS = [
        "item_id", "title", "subject", "price", "price_unit",
        "quantity", "min_buy_quantity", "step_price", "mix_enable",
        "pic_url", "detail_url", "category_id", "trade_props",
        "sku_infos", "seller"
    ]
    
    # 调用接口获取商品详情
    print(f"获取商品 {TARGET_ITEM_ID} 详情...")
    raw_item = item_api.get_item_detail(
        item_id=TARGET_ITEM_ID,
        fields=REQUIRED_FIELDS    )
    
    # 解析并展示结果
    if raw_item:
        item_detail = parse_b2b_item_detail(raw_item)
        
        print("\n===== 商品基础信息 =====")
        for key, value in item_detail["base_info"].items():
            print(f"{key}: {value}")
            
        print("\n===== 批发属性 =====")
        print(f"起订量: {item_detail['wholesale_info']['min_buy_quantity']}{item_detail['base_info']['price_unit']}")
        print(f"是否支持混批: {'是' if item_detail['wholesale_info']['can_mix'] else '否'}")
        print("阶梯价格:")
        for step in item_detail['wholesale_info']['step_price']:
            print(f"- {step['quantity']}件及以上: {step['price']}元/件")
            
        print("\n===== 供应商信息 =====")
        for key, value in item_detail["supplier_info"].items():
            print(f"{key}: {value}")
            
        # 处理SKU数据
        sku_df = sku_to_dataframe(item_detail["sku_list"])
        if not sku_df.empty:
            print("\n===== SKU信息 =====")
            print(sku_df[["sku_id", "price", "quantity", "颜色", "尺码"]].to_string(index=False))
            
            # 保存SKU数据到CSV
            sku_df.to_csv(f"1688_sku_{TARGET_ITEM_ID}.csv", index=False, encoding="utf-8-sig")
            print(f"\nSKU数据已保存至 1688_sku_{TARGET_ITEM_ID}.csv")

def calculate_wholesale_price(step_prices: list, quantity: int) -> str:
    """
    根据采购数量计算批发价格
    :param step_prices: 阶梯价格列表
    :param quantity: 采购数量
    :return: 对应价格
    """
    if not step_prices:
        return "0.00"
        
    # 按数量排序（升序）
    sorted_steps = sorted(step_prices, key=lambda x: int(x["quantity"]))
    
    # 找到适用的价格阶梯
    applicable_price = sorted_steps[0]["price"]
    for step in sorted_steps:
        if quantity >= int(step["quantity"]):
            applicable_price = step["price"]
        else:
            break
            
    return applicable_price# 使用示例if item_detail and item_detail["wholesale_info"]["step_price"]:
    order_quantity = 200
    price = calculate_wholesale_price(
        item_detail["wholesale_info"]["step_price"], 
        order_quantity    )
    print(f"\n采购{order_quantity}件的单价为: {price}元")

def filter_supplier(supplier_info: Dict, min_trust_year: int = 3, min_credit: int = 3) -> bool:
    """
    筛选优质供应商
    :param supplier_info: 供应商信息
    :param min_trust_year: 最低诚信通年限
    :param min_credit: 最低信用等级
    :return: 是否符合条件
    """
    if not supplier_info.get("is_verified"):
        return False
        
    if int(supplier_info.get("trust_year", 0)) < min_trust_year:
        return False
        
    if int(supplier_info.get("credit_level", 0)) < min_credit:
        return False
        
    return True

import redisimport picklefrom datetime import timedeltaclass CachedAlibabaAPI(AlibabaItemAPI):
    def __init__(self, app_key, app_secret, redis_host="localhost", redis_port=6379):
        super().__init__(app_key, app_secret)
        self.redis = redis.Redis(host=redis_host, port=redis_port, db=0)
        
    def get_cached_item_detail(self, item_id, expire=3600, **kwargs):
        """带缓存的商品详情获取"""
        cache_key = f"1688_item_{item_id}"
        
        # 尝试从缓存获取
        cached_data = self.redis.get(cache_key)
        if cached_data:
            return pickle.loads(cached_data)
            
        # 调用接口
        item_detail = self.get_item_detail(item_id,** kwargs)
        
        # 存入缓存
        if item_detail:
            self.redis.setex(cache_key, timedelta(seconds=expire), pickle.dumps(item_detail))
            
        return item_detail

from tenacity import retry, stop_after_attempt, wait_exponentialclass ReliableAlibabaAPI(AlibabaItemAPI):
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def get_reliable_item_detail(self, *args, **kwargs):
        """带重试机制的商品详情获取"""
        return super().get_item_detail(*args, **kwargs)

def safe_get_item_detail(api_client, item_id, max_retries=3):
    """安全获取商品详情，带重试和详细日志"""
    retries = 0
    while retries < max_retries:
        try:
            return api_client.get_item_detail(item_id)
        except Exception as e:
            retries += 1
            print(f"第{retries}次重试获取商品{item_id}详情: {str(e)}")
            if retries >= max_retries:
                # 记录错误日志到文件
                with open("error_log.txt", "a") as f:
                    f.write(f"获取商品{item_id}详情失败: {str(e)} - {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                return None
            time.sleep(2 ** retries)  # 指数退避