唯品会作为国内头部品牌特卖电商平台，其商品详情接口（开放平台稳定版：/api/product/detail/v2.0）是企业级应用（竞品监控、价格跟踪、选品分析、供应链联动）获取商品核心数据的关键入口。不同于综合电商的商品详情接口，唯品会接口深度绑定“品牌特卖、限时折扣、多规格分层价”的核心模式，存在“时效敏感、库存动态变化、促销规则复杂、地区差异化、签名机制严苛”等特色痛点。当前全网技术贴均停留在“抓包获取临时接口+简单字段提取”的浅层层面，既忽视唯品会开放平台的合规约束（非法抓包易导致账号封禁），也未解决生产环境中“特卖时效过期、多规格库存不联动、促销规则解析混乱、异常响应无兜底”等实际问题；同时，与我之前撰写的VVIC图片/关键词搜索接口贴文相比，本次完全摒弃“预处理+调度”的通用框架，聚焦唯品会特卖场景的专属需求，打造“合规对接+场景适配+高可用兜底”的全流程方案，所有代码可直接落地企业级生产环境，兼顾合规性与业务价值，完全适配CSDN技术贴规范，无任何全网同质化内容

一、核心认知：唯品会商品详情接口的差异化特性（区别于全网+过往贴文）

唯品会商品详情接口与综合电商（淘宝、京东）、自身非特卖接口（如商品列表）、以及我之前对接的VVIC接口差异显著，其设计逻辑完全围绕“品牌特卖”场景展开，四大核心特性直接决定对接思路——照搬通用接口对接经验、复用非特卖场景框架，必然导致合规风险、数据失真、业务适配性差，这也是全网现有教程的核心盲区：

1. 合规约束严苛，禁止非法抓包：唯品会开放平台对接口调用有明确的权限分级（个人开发者/企业开发者），仅企业开发者可获取完整的商品详情数据（含库存、促销、历史价格）；全网现有教程常用的“移动端抓包临时接口”（如/mapi/vip.com/rest/product/detail/info）属于平台内部接口，非法调用会触发IP封禁、账号拉黑，且接口地址动态变化，无法用于生产环境[3][4]。本次方案全程基于唯品会开放平台官方接口对接，严格遵循《唯品会开放平台服务条款》，杜绝合规风险。

2. 特卖时效敏感，数据动态性极强：商品详情中核心数据（折扣价、库存、促销活动）均带有时效性（如限时特卖24小时、库存实时售罄），接口返回的“特卖开始时间、结束时间”需精准解析，否则会出现“获取到过期折扣价、显示有库存但无法下单”的问题；且同一商品在不同时间段（特卖前、特卖中、特卖后）的返回字段结构会发生变化，需适配多场景字段解析[2][4]。

3. 多规格与库存、价格深度联动：不同于常规电商“多规格同价”的模式，唯品会商品常存在“多规格分层价”（如不同尺码、颜色对应不同折扣价），且各规格库存独立、地区库存差异明显（如北京有库存、上海无库存）；同时，规格选择会影响促销规则适配（如某尺码不参与满减），常规字段提取会导致“价格与规格不匹配、库存显示错误”[2][3]。

4. 促销规则复杂，解析难度高：商品详情接口返回的促销信息包含“限时折扣、满减、优惠券、会员价、品牌补贴”等多种形式，且存在促销叠加规则（如满减+会员价可同时生效）、促销排斥规则（如优惠券与品牌补贴二选一）；全网现有教程仅能提取促销标签，无法解析具体规则，导致业务无法落地（如无法判断实际到手价）[2][4]。

核心提醒：1. 本文方案全程基于唯品会开放平台官方接口开发，需提前注册企业开发者账号、创建应用并通过审核，获取appKey和appSecret（个人开发者仅能获取基础商品信息，无库存、促销等核心字段）；2. 接口调用需严格遵守开放平台限流规则（基础权限20次/分钟、高级权限100次/分钟，按appKey+IP双重限制），避免触发风控；3. 与我过往撰写的VVIC接口贴文相比，本次无任何模块复用，聚焦唯品会特卖场景专属需求，重点解决“合规对接、时效适配、多规格联动、促销解析、异常兜底”五大核心问题，与全网基础教程形成本质区别；4. 接口签名采用HMAC-SHA256算法，与VVIC的MD5签名逻辑完全不同，需单独适配，避免签名失败。

点击获取key和secret

二、差异化方案实现：五大核心模块（全特卖场景专属，无过往模块复用）

方案基于唯品会开放平台V2.0商品详情接口构建，核心包含“合规签名客户端+特卖时效适配模块+多规格库存联动解析器+促销规则深度解析模块+异常熔断兜底架构”，技术栈以Python为主，兼顾合规性、场景适配性与高可用性，全程围绕唯品会“品牌特卖”核心，每一个模块均为全网现有教程未涉及的进阶内容，彻底摆脱同质化困境。

1. 合规签名客户端：解决签名失败与合规调用问题

这是唯品会接口对接的基础前提，也是全网现有教程最易忽视的核心环节。全网教程多采用“抓包获取临时接口+跳过签名”的非法方式，无法用于生产环境；而官方接口的签名机制（HMAC-SHA256）存在“参数排序严格、时间戳偏差限制、密钥加密规范”等隐蔽要求，任一环节错误都会返回“401签名无效”。本客户端针对唯品会开放平台规范，实现“合规签名生成+请求频率控制+参数校验+权限适配”全流程，确保接口调用合规、稳定，避免签名失败与风控触发[4]：

import hashlib import hmac import time import json import requests from typing import Dict, Optional, Any from datetime import datetime, timedelta from threading import Lock class VipshopComplianceClient: """唯品会商品详情接口合规签名客户端：适配开放平台V2.0，解决签名失败与合规调用问题""" def __init__(self, app_key: str, app_secret: str, timeout: int = 10, max_calls_per_minute: int = 20): self.app_key = app_key # 开放平台申请的appKey self.app_secret = app_secret.encode("utf-8") # 密钥，需编码为字节流 self.base_url = "https://api.vip.com/product/detail/v2.0" # 官方稳定接口地址 self.timeout = timeout # 请求超时时间（秒） self.max_calls = max_calls_per_minute # 限流阈值（按开放平台权限配置） self.last_request_time = 0 # 上一次请求时间，用于频率控制 self.request_lock = Lock() # 线程锁，保证多线程下频率控制安全 # 接口必传公共参数（V2.0版本强制要求） self.public_params = { "appKey": self.app_key, "format": "json", "v": "2.0", "signMethod": "HmacSHA256", "timestamp": "", # 动态生成毫秒级时间戳 "region": "110000" # 默认地区编码（北京），可动态调整 } def _generate_sign(self, params: Dict) -> str: """生成唯品会开放平台合规签名（HMAC-SHA256），严格遵循官方规则[4]""" # 1. 排除sign字段，按参数名ASCII码升序排序（核心：排序错误直接签名失败） sorted_params = sorted([(k, v) for k, v in params.items() if k != "sign"], key=lambda x: x[0]) # 2. 拼接为"key=value&key=value"格式，中文参数需UTF-8编码 sign_str = "&".join([f"{k}={self._encode_chinese(v)}" for k, v in sorted_params]) # 3. 用appSecret作为密钥，进行HMAC-SHA256加密，结果Base64编码（核心区别于MD5签名） hmac_obj = hmac.new(self.app_secret, sign_str.encode("utf-8"), hashlib.sha256) sign = hmac_obj.digest().hex() # 转为16进制字符串，无需Base64编码（V2.0版本规范） return sign.lower() def _encode_chinese(self, value: Any) -> str: """中文参数编码：适配唯品会接口要求，中文需UTF-8编码，避免乱码与签名失败""" if isinstance(value, str) and any("\u4e00" <= char <= "\u9fa5" for char in value): return value.encode("utf-8").decode("utf-8") return str(value) def _control_request_freq(self): """请求频率控制：基于限流阈值，控制请求间隔，避免触发风控[2]""" with self.request_lock: current_time = time.time() # 计算需等待的时间（确保1分钟内请求不超过阈值） interval = 60 / self.max_calls time_diff = current_time - self.last_request_time if time_diff < interval: time.sleep(interval - time_diff) self.last_request_time = time.time() def _validate_params(self, product_id: str, region: Optional[str] = None) -> None: """参数校验：避免无效参数导致接口调用失败（全网教程未涉及）""" if not product_id or not product_id.isdigit(): raise ValueError("商品ID无效：必须为纯数字（从商品详情页URL提取，如detail-12345678.html中的12345678）[3]") if region and (not region.isdigit() or len(region) != 6): raise ValueError("地区编码无效：必须为6位数字（如北京110000、上海310000）") def get_raw_detail(self, product_id: str, region: Optional[str] = None) -> Dict: """ 合规调用商品详情接口，获取原始响应数据（无任何字段过滤，为后续解析提供基础） :param product_id: 商品唯一ID（纯数字，从商品详情页URL提取）[3] :param region: 地区编码（可选，影响地区库存和价格） :return: 接口原始响应JSON """ # 1. 参数校验 self._validate_params(product_id, region) # 2. 频率控制 self._control_request_freq() # 3. 拼接请求参数（公共参数+业务参数） request_params = self.public_params.copy() request_params["timestamp"] = str(int(time.time() * 1000)) # 毫秒级时间戳，偏差≤3分钟[4] if region: request_params["region"] = region # 业务参数（必传：商品ID） request_params["productId"] = product_id # 4. 生成签名 request_params["sign"] = self._generate_sign(request_params) try: # 5. 发送POST请求（V2.0版本强制要求POST，GET请求会返回405错误） response = requests.post( url=self.base_url, data=json.dumps(request_params), headers={"Content-Type": "application/json;charset=utf-8"}, timeout=self.timeout, verify=True # 开启SSL证书验证，避免安全风险（全网教程多关闭，存在隐患） ) response.raise_for_status() # 触发HTTP错误（如401、429）时抛出异常 return response.json() except requests.exceptions.RequestException as e: # 捕获请求异常，返回标准化错误信息 return {"code": 500, "msg": f"接口调用异常：{str(e)}", "data": None} except json.JSONDecodeError: return {"code": 500, "msg": "接口响应解析失败（非JSON格式）", "data": None} # 示例：合规调用接口获取原始数据 if __name__ == "__main__": # 替换为自己的开放平台appKey和appSecret（企业开发者账号获取） CLIENT = VipshopComplianceClient( app_key="YOUR_APP_KEY", app_secret="YOUR_APP_SECRET", max_calls_per_minute=20 # 基础权限，20次/分钟 ) # 示例商品ID（从唯品会商品详情页URL提取） product_id = "12345678" # 调用接口（指定地区为上海，310000） raw_response = CLIENT.get_raw_detail(product_id=product_id, region="310000") print(f"接口调用状态：{'成功' if raw_response.get('code') == 0 else '失败'}") if raw_response.get("code") == 0: print(f"商品基础信息：{raw_response['data']['productInfo']['productName']}") else: print(f"错误信息：{raw_response.get('msg')}")

2. 特卖时效适配模块：解决时效过期与数据动态变化问题

这是本次贴文的核心差异化亮点之一，全网现有教程均未涉及。唯品会商品的核心特性是“限时特卖”，接口返回的商品数据（折扣价、库存、促销）均有明确的时效范围，且不同特卖阶段（未开始、进行中、已结束）的返回字段结构不同——若未适配时效，会出现“获取到过期折扣价、显示有库存但无法下单、特卖结束后仍展示促销标签”等业务问题。本模块基于接口返回的时效字段，实现“时效判断+多阶段适配+过期兜底”，确保数据与商品实际状态一致，适配特卖场景的动态需求[2][3]：

from typing import Dict, Optional, Any from datetime import datetime from vipshop_compliance_client import VipshopComplianceClient class VipshopSaleTimeAdapter: """唯品会特卖时效适配模块：时效判断+多阶段适配+过期兜底，解决数据动态变化问题""" def __init__(self, client: VipshopComplianceClient): self.client = client # 特卖阶段映射（用于标准化阶段标识，适配不同字段返回） self.sale_stage_map = { 0: "NOT_STARTED", # 特卖未开始 1: "ONGOING", # 特卖进行中 2: "ENDED", # 特卖已结束 3: "SELL_OUT" # 特卖中但已售罄 } def _parse_sale_time(self, product_data: Dict) -> Dict: """解析特卖时效字段：提取开始时间、结束时间，转换为标准化格式[3]""" sale_info = product_data.get("saleInfo", {}) # 提取接口返回的毫秒级时间戳，转换为datetime格式 start_time_ms = sale_info.get("saleStartTime", 0) end_time_ms = sale_info.get("saleEndTime", 0) # 时间戳转换（处理0值异常，避免报错） start_time = datetime.fromtimestamp(start_time_ms / 1000) if start_time_ms != 0 else None end_time = datetime.fromtimestamp(end_time_ms / 1000) if end_time_ms != 0 else None current_time = datetime.now() # 判断特卖阶段 if not start_time or not end_time: stage = self.sale_stage_map[2] # 无时效信息，默认视为已结束 elif current_time < start_time: stage = self.sale_stage_map[0] # 未开始 elif current_time<= end_time: # 特卖进行中，判断是否售罄 total_stock = product_data.get("stockInfo", {}).get("totalStock", 0) stage = self.sale_stage_map[3] if total_stock <= 0 else self.sale_stage_map[1] else: stage = self.sale_stage_map[2] # 已结束 return { "sale_stage": stage, "sale_start_time": start_time.strftime("%Y-%m-%d %H:%M:%S") if start_time else None, "sale_end_time": end_time.strftime("%Y-%m-%d %H:%M:%S") if end_time else None, "remaining_time": self._calculate_remaining_time(end_time) if end_time and stage == self.sale_stage_map[1] else None } def _calculate_remaining_time(self, end_time: datetime) -> str: """计算特卖剩余时间（仅特卖进行中有效），格式化展示""" remaining = end_time - datetime.now() if remaining.total_seconds() <= 0: return "00:00:00" hours, remainder = divmod(remaining.total_seconds(), 3600) minutes, seconds = divmod(remainder, 60) return f"{int(hours):02d}:{int(minutes):02d}:{int(seconds):02d}" def _adapt_stage_fields(self, raw_data: Dict) -> Dict: """多阶段字段适配：不同特卖阶段，提取对应核心字段，避免字段缺失报错[2]""" if raw_data.get("code") != 0 or not raw_data.get("data"): return {"code": raw_data.get("code"), "msg": raw_data.get("msg"), "data": None} product_data = raw_data["data"] sale_time_info = self._parse_sale_time(product_data) adapted_data = {"sale_time_info": sale_time_info} # 不同阶段，提取不同核心字段 if sale_time_info["sale_stage"] == self.sale_stage_map[0]: # 特卖未开始：提取基础信息、预售价、开始时间 adapted_data["base_info"] = self._extract_base_info(product_data) adapted_data["pre_sale_info"] = self._extract_pre_sale_info(product_data) elif sale_time_info["sale_stage"] in [self.sale_stage_map[1], self.sale_stage_map[3]]: # 特卖进行中/售罄：提取基础信息、当前折扣价、库存、促销信息 adapted_data["base_info"] = self._extract_base_info(product_data) adapted_data["price_info"] = self._extract_price_info(product_data) adapted_data["stock_info"] = self._extract_stock_info(product_data) else: # 特卖已结束：提取基础信息、历史折扣价、恢复原价 adapted_data["base_info"] = self._extract_base_info(product_data) adapted_data["historical_price_info"] = self._extract_historical_price_info(product_data) return {"code": 0, "msg": "success", "data": adapted_data} def _extract_base_info(self, product_data: Dict) -> Dict: """提取商品基础信息（所有阶段通用）[3]""" product_info = product_data.get("productInfo", {}) brand_info = product_data.get("brandInfo", {}) return { "product_id": product_info.get("productId", ""), "product_name": product_info.get("productName", ""), "brand_id": brand_info.get("brandId", ""), "brand_name": brand_info.get("brandName", ""), "category_path": product_info.get("categoryPath", ""), # 类目路径，如"运动户外>运动鞋" "main_images": product_info.get("mainImages", []), # 主图URL列表 "detail_url": f"https://detail.vip.com/detail-{product_info.get('brandId', '')}-{product_info.get('productId', '')}.html" # 商品详情页URL[3] } def _extract_pre_sale_info(self, product_data: Dict) -> Dict: """提取预售价信息（特卖未开始阶段）""" pre_sale_info = product_data.get("preSaleInfo", {}) return { "pre_sale_price": pre_sale_info.get("preSalePrice", 0.0), "original_price": pre_sale_info.get("originalPrice", 0.0), "discount": round(pre_sale_info.get("preSalePrice", 0.0) / pre_sale_info.get("originalPrice", 1.0), 2) if pre_sale_info.get("originalPrice", 0.0) != 0 else 0.0 } def _extract_price_info(self, product_data: Dict) -> Dict: """提取当前价格信息（特卖进行中阶段）[2]""" price_info = product_data.get("priceInfo", {}) return { "current_price": price_info.get("currentPrice", 0.0), # 当前折扣价 "original_price": price_info.get("originalPrice", 0.0), # 原价 "discount": price_info.get("discount", 0.0), # 折扣率（如0.3表示3折） "member_price": price_info.get("memberPrice", 0.0), # 会员价（SVIP专享） "price_note": price_info.get("priceNote", "") # 价格说明（如"限时直降"） } def _extract_historical_price_info(self, product_data: Dict) -> Dict: """提取历史价格信息（特卖已结束阶段）""" historical_price_info = product_data.get("historicalPriceInfo", {}) return { "highest_price": historical_price_info.get("highestPrice", 0.0), "lowest_price": historical_price_info.get("lowestPrice", 0.0), "average_price": historical_price_info.get("averagePrice", 0.0), "restore_price": historical_price_info.get("restorePrice", 0.0) # 恢复原价 } def _extract_stock_info(self, product_data: Dict) -> Dict: """提取库存信息（特卖进行中阶段，基础库存，多规格库存后续单独解析）""" stock_info = product_data.get("stockInfo", {}) return { "total_stock": stock_info.get("totalStock", 0), # 总库存 "sell_out": stock_info.get("totalStock", 0) <= 0, # 是否售罄 "region_stock_note": stock_info.get("regionStockNote", "") # 地区库存说明（如"上海无库存"） } def get_adapted_detail(self, product_id: str, region: Optional[str] = None) -> Dict: """ 时效适配后的商品详情：获取原始数据→时效判断→多阶段字段适配 :param product_id: 商品ID :param region: 地区编码 :return: 时效适配后的标准化商品详情数据 """ raw_data = self.client.get_raw_detail(product_id=product_id, region=region) return self._adapt_stage_fields(raw_data) # 示例：时效适配获取商品详情 if __name__ == "__main__": CLIENT = VipshopComplianceClient( app_key="YOUR_APP_KEY", app_secret="YOUR_APP_SECRET" ) ADAPTER = VipshopSaleTimeAdapter(client=CLIENT) # 调用时效适配后的接口 adapted_detail = ADAPTER.get_adapted_detail(product_id="12345678", region="310000") if adapted_detail["code"] == 0: print(f"特卖阶段：{adapted_detail['data']['sale_time_info']['sale_stage']}") print(f"特卖时间：{adapted_detail['data']['sale_time_info']['sale_start_time']} - {adapted_detail['data']['sale_time_info']['sale_end_time']}") if adapted_detail['data']['sale_time_info']['sale_stage'] == "ONGOING": print(f"剩余时间：{adapted_detail['data']['sale_time_info']['remaining_time']}") print(f"当前价格：{adapted_detail['data']['price_info']['current_price']}元（{adapted_detail['data']['price_info']['discount']}折）") else: print(f"商品基础信息：{adapted_detail['data']['base_info']['product_name']}（{adapted_detail['data']['base_info']['brand_name']}）") else: print(f"接口调用失败：{adapted_detail['msg']}")

3. 多规格库存联动解析器：解决规格与价格、库存不匹配问题

这是全网现有教程均未涉及的核心模块，也是唯品会特卖场景的刚需。唯品会商品常存在“多规格分层价”（如不同尺码、颜色对应不同折扣价、不同库存），且各规格独立库存、地区库存差异明显——全网现有教程仅能提取总库存和基础价格，无法关联“规格-价格-库存”三者关系，导致业务无法落地（如无法判断某一规格的实际价格和库存）。本解析器针对唯品会多规格特性，实现“规格结构解析+三者联动关联+地区库存适配+异常规格过滤”，自动关联每一个规格对应的价格、库存、地区可用性，输出标准化的联动数据，直接支撑选品、下单等业务需求[2][3]：

from typing import Dict, List, Optional, Tuple from vipshop_sale_time_adapter import VipshopSaleTimeAdapter class VipshopSpecStockLinkParser: """唯品会多规格库存联动解析器：关联规格-价格-库存，解决三者不匹配问题""" def __init__(self, adapter: VipshopSaleTimeAdapter): self.adapter = adapter # 规格类型映射（标准化规格名称，适配不同商品的规格表述差异） self.spec_type_map = { "颜色": "color", "尺码": "size", "版型": "version", "材质": "material" } def _parse_spec_structure(self, product_data: Dict) -> List[Dict]: """解析多规格结构：提取规格组、规格项，标准化规格类型[3]""" spec_info = product_data.get("specInfo", {}) spec_groups = spec_info.get("specGroups", []) parsed_specs = [] for group in spec_groups: spec_type = group.get("specTypeName", "") # 标准化规格类型（如"颜色"转为"color"） standard_type = self.spec_type_map.get(spec_type, spec_type.lower()) spec_items = group.get("specItems", []) parsed_items = [] for item in spec_items: # 解析单个规格项的基础信息 parsed_items.append({ "spec_item_id": item.get("specItemId", ""), # 规格项唯一ID "spec_item_name": item.get("specItemName", ""), # 规格项名称（如"黑色"、"42码"） "spec_item_image": item.get("specItemImage", ""), # 规格项图片（如颜色图） "is_available": item.get("isAvailable", False) # 该规格是否可用 }) parsed_specs.append({ "spec_type": standard_type, "spec_type_name": spec_type, "spec_items": parsed_items }) return parsed_specs def _link_spec_price_stock(self, product_data: Dict, parsed_specs: List[Dict]) -> List[Dict]: """联动规格-价格-库存：关联每一个规格项对应的价格和库存[2]""" # 提取规格价格映射（不同规格对应不同价格） spec_price_map = self._get_spec_price_map(product_data) # 提取规格库存映射（不同规格对应不同库存） spec_stock_map = self._get_spec_stock_map(product_data) # 提取地区编码（用于判断地区库存可用性） region = product_data.get("requestParams", {}).get("region", "110000") # 联动规格、价格、库存 for spec_group in parsed_specs: for spec_item in spec_group["spec_items"]: spec_item_id = spec_item["spec_item_id"] # 关联价格（无对应价格则使用基础价格） spec_price = spec_price_map.get(spec_item_id, {}) spec_item["price"] = spec_price.get("currentPrice", 0.0) spec_item["original_price"] = spec_price.get("originalPrice", 0.0) spec_item["discount"] = spec_price.get("discount", 0.0) # 关联库存（无对应库存则视为0） spec_stock = spec_stock_map.get(spec_item_id, {}) spec_item["stock"] = spec_stock.get("stock", 0) spec_item["region_available"] = spec_stock.get("regionAvailable", {}).get(region, False) # 地区库存可用性 # 补充规格可用性（结合库存和地区可用性） spec_item["is_available"] = spec_item["is_available"] and spec_item["stock"] > 0 and spec_item["region_available"] return parsed_specs def _get_spec_price_map(self, product_data: Dict) -> Dict: """获取规格-价格映射：提取每一个规格项对应的价格信息""" spec_price_info = product_data.get("specPriceInfo", {}) spec_price_list = spec_price_info.get("specPriceList", []) spec_price_map = {} for spec_price in spec_price_list: spec_item_id = spec_price.get("specItemId", "") if spec_item_id: spec_price_map[spec_item_id] = { "currentPrice": spec_price.get("currentPrice", 0.0), "originalPrice": spec_price.get("originalPrice", 0.0), "discount": spec_price.get("discount", 0.0) } return spec_price_map def _get_spec_stock_map(self, product_data: Dict) -> Dict: """获取规格-库存映射：提取每一个规格项对应的库存信息和地区可用性[3]""" spec_stock_info = product_data.get("specStockInfo", {}) spec_stock_list = spec_stock_info.get("specStockList", []) spec_stock_map = {} for spec_stock in spec_stock_list: spec_item_id = spec_stock.get("specItemId", "") if spec_item_id: # 解析地区库存可用性（如{"110000": true, "310000": false}） region_available = {} region_stock_list = spec_stock.get("regionStockList", []) for region_stock in region_stock_list: region_code = region_stock.get("region", "") region_available[region_code] = region_stock.get("available", False) spec_stock_map[spec_item_id] = { "stock": spec_stock.get("stock", 0), "regionAvailable": region_available } return spec_stock_map def _filter_invalid_specs(self, linked_specs: List[Dict]) -> List[Dict]: """过滤无效规格：移除不可用、无价格、无库存的规格项，简化数据结构""" filtered_specs = [] for spec_group in linked_specs: valid_items = [item for item in spec_group["spec_items"] if item["is_available"]] if valid_items: spec_group["spec_items"] = valid_items filtered_specs.append(spec_group) return filtered_specs def get_spec_stock_link_detail(self, product_id: str, region: Optional[str] = None) -> Dict: """ 全流程规格-价格-库存联动解析：获取时效适配数据→解析规格结构→联动三者→过滤无效规格 :param product_id: 商品ID :param region: 地区编码 :return: 标准化的规格-价格-库存联动数据 """ # 1. 获取时效适配后的商品详情数据 adapted_detail = self.adapter.get_adapted_detail(product_id=product_id, region=region) if adapted_detail.get("code") != 0 or not adapted_detail.get("data"): return adapted_detail product_data = adapted_detail["data"] # 仅特卖进行中，解析规格-价格-库存联动数据（其他阶段无意义） if product_data["sale_time_info"]["sale_stage"] != "ONGOING": product_data["spec_stock_link_info"] = {"msg": "特卖未开始/已结束，无需解析规格库存联动数据", "data": []} return adapted_detail # 2. 解析规格结构 parsed_specs = self._parse_spec_structure(adapted_detail["data"]) if not parsed_specs: product_data["spec_stock_link_info"] = {"msg": "该商品无多规格", "data": []} return adapted_detail # 3. 联动规格-价格-库存 linked_specs = self._link_spec_price_stock(adapted_detail["data"], parsed_specs) # 4. 过滤无效规格 filtered_specs = self._filter_invalid_specs(linked_specs) # 补充规格库存联动信息 product_data["spec_stock_link_info"] = { "msg": "success", "has_multi_spec": len(filtered_specs) > 0, "spec_groups": filtered_specs, "valid_spec_count": sum([len(group["spec_items"]) for group in filtered_specs]) } return adapted_detail # 示例：规格-价格-库存联动解析 if __name__ == "__main__": CLIENT = VipshopComplianceClient( app_key="YOUR_APP_KEY", app_secret="YOUR_APP_SECRET" ) ADAPTER = VipshopSaleTimeAdapter(client=CLIENT) PARSER = VipshopSpecStockLinkParser(adapter=ADAPTER) # 调用联动解析接口（以上海地区为例，310000） link_detail = PARSER.get_spec_stock_link_detail(product_id="12345678", region="310000") if link_detail["code"] == 0: sale_stage = link_detail["data"]["sale_time_info"]["sale_stage"] if sale_stage == "ONGOING": spec_link_info = link_detail["data"]["spec_stock_link_info"] print(f"商品是否有多规格：{'是' if spec_link_info['has_multi_spec'] else '否'}") print(f"有效规格项数量：{spec_link_info['valid_spec_count']}") # 打印每一个规格组的联动信息 for spec_group in spec_link_info["spec_groups"]: print(f"\n规格类型：{spec_group['spec_type_name']}") for spec_item in spec_group["spec_items"]: print(f" 规格：{spec_item['spec_item_name']}，价格：{spec_item['price']}元（{spec_item['discount']}折），库存：{spec_item['stock']}件，上海地区可用：{spec_item['region_available']}") else: print(f"特卖阶段：{sale_stage}，无需解析规格库存联动数据") else: print(f"接口调用失败：{link_detail['msg']}")

4. 促销规则深度解析模块：解决促销规则混乱、无法落地问题

这是唯品会特卖场景的核心难点，也是全网现有教程的最大盲区。唯品会商品详情接口返回的促销信息包含“限时折扣、满减、优惠券、会员价、品牌补贴”等多种形式，且存在复杂的叠加/排斥规则（如满减+会员价可叠加，优惠券与品牌补贴二选一）——全网现有教程仅能提取促销标签（如“满300减50”），无法解析具体规则和叠加逻辑，导致业务无法判断实际到手价、无法适配促销场景。本模块针对唯品会促销特性，实现“促销类型分类+规则深度解析+叠加逻辑判断+到手价计算”，输出标准化的促销规则和实际到手价，直接支撑价格监控、选品分析等业务需求[2][4]：

from typing import Dict, List, Optional, Tuple from vipshop_spec_stock_link_parser import VipshopSpecStockLinkParser class VipshopPromotionRuleParser: """唯品会促销规则深度解析模块：解析促销规则+判断叠加逻辑+计算到手价""" def __init__(self, parser: VipshopSpecStockLinkParser): self.parser = parser # 促销类型映射（标准化促销类型，适配接口返回的多种表述） self.promotion_type_map = { "LIMITED_DISCOUNT": "限时折扣", "FULL_REDUCE": "满减", "COUPON": "优惠券", "MEMBER_PRICE": "会员价", "BRAND_SUBSIDY": "品牌补贴", "FULL_GIFT": "满赠" } # 促销叠加规则（key: 促销类型，value: 可叠加的促销类型列表） self.promotion_stack_rule = { "LIMITED_DISCOUNT": ["FULL_REDUCE", "MEMBER_PRICE", "BRAND_SUBSIDY"], "FULL_REDUCE": ["LIMITED_DISCOUNT", "MEMBER_PRICE"], "MEMBER_PRICE": ["LIMITED_DISCOUNT", "FULL_REDUCE"], "COUPON": ["LIMITED_DISCOUNT"], # 优惠券仅可与限时折扣叠加 "BRAND_SUBSIDY": ["LIMITED_DISCOUNT", "MEMBER_PRICE"], "FULL_GIFT": [] # 满赠不与任何促销叠加 } def _classify_promotions(self, product_data: Dict) -> List[Dict]: """促销类型分类：提取所有促销信息，分类并标准化[2][4]""" promotion_info = product_data.get("promotionInfo", {}) promotions = promotion_info.get("promotions", []) classified_promotions = [] for promo in promotions: promo_type = promo.get("promotionType", "") # 标准化促销类型 standard_type = next((k for k, v in self.promotion_type_map.items() if v == promo_type), promo_type) # 提取核心促销信息，适配不同促销类型的字段差异 if standard_type == "LIMITED_DISCOUNT": promo_detail = self._parse_limited_discount(promo) elif standard_type == "FULL_REDUCE": promo_detail = self._parse_full_reduce(promo) elif standard_type == "COUPON": promo_detail = self._parse_coupon(promo) elif standard_type == "MEMBER_PRICE": promo_detail = self._parse_member_price(promo) elif standard_type == "BRAND_SUBSIDY": promo_detail = self._parse_brand_subsidy(promo) elif standard_type == "FULL_GIFT": promo_detail = self._parse_full_gift(promo) else: promo_detail = {"promotion_desc": promo.get("promotionDesc", "")} classified_promotions.append({ "promotion_id": promo.get("promotionId", ""), "promotion_type": standard_type, "promotion_type_name": self.promotion_type_map.get(standard_type, standard_type), "promotion_desc": promo.get("promotionDesc", ""), "is_valid": promo.get("isValid", False), "start_time": promo.get("startTime", ""), "end_time": promo.get("endTime", ""), "detail": promo_detail }) # 过滤无效促销 return [promo for promo in classified_promotions if promo["is_valid"]] def _parse_limited_discount(self, promo: Dict) -> Dict: """解析限时折扣规则""" return { "discount_rate": promo.get("discountRate", 0.0), # 折扣率 "discount_price": promo.get("discountPrice", 0.0), # 折扣价 "original_price": promo.get("originalPrice", 0.0) # 原价 } def _parse_full_reduce(self, promo: Dict) -> Dict: """解析满减规则（支持多档位满减，如满300减50、满500减100）[4]""" full_reduce_details = promo.get("fullReduceDetails", []) stages = [] for stage in full_reduce_details: stages.append({ "full_amount": stage.get("fullAmount", 0.0), # 满多少 "reduce_amount": stage.get("reduceAmount", 0.0), # 减多少 "max_reduce": stage.get("maxReduce", 0.0) # 单次最大减免 }) return {"stages": stages} def _parse_coupon(self, promo: Dict) -> Dict: """解析优惠券规则""" return { "coupon_id": promo.get("couponId", ""), "denomination": promo.get("denomination", 0.0), # 优惠券面额 "full_amount": promo.get("fullAmount", 0.0), # 使用门槛（满多少可用） "usable_range": promo.get("usableRange", "") # 使用范围（如"全品类可用"） } def _parse_member_price(self, promo: Dict) -> Dict: """解析会员价规则[2]""" return { "member_price": promo.get("memberPrice", 0.0), # 会员价 "original_price": promo.get("originalPrice", 0.0), # 原价 "discount_rate": round(promo.get("memberPrice", 0.0) / promo.get("originalPrice", 1.0), 2) if promo.get("originalPrice", 0.0) != 0 else 0.0 } def _parse_brand_subsidy(self, promo: Dict) -> Dict: """解析品牌补贴规则""" return { "subsidy_amount": promo.get("subsidyAmount", 0.0), # 补贴金额 "subsidy_price": promo.get("subsidyPrice", 0.0), # 补贴后价格 "original_price": promo.get("originalPrice", 0.0) # 原价 } def _parse_full_gift(self, promo: Dict) -> Dict: """解析满赠规则""" return { "full_amount": promo.get("fullAmount", 0.0), # 满多少赠 "gift_name": promo.get("giftName", ""), # 赠品名称 "gift_count": promo.get("giftCount", 1) # 赠品数量 } def _judge_stack_logic(self, classified_promotions: List[Dict]) -> List[Dict]: """判断促销叠加逻辑：标记每一个促销可叠加的促销列表[4]""" for promo in classified_promotions: usable_stack_types = self.promotion_stack_rule.get(promo["promotion_type"], []) # 筛选可叠加的促销ID stack_promo_ids = [ p["promotion_id"] for p in classified_promotions if p["promotion_type"] in usable_stack_types and p["promotion_id"] != promo["promotion_id"] ] promo["stack_promo_ids"] = stack_promo_ids promo["stack_promo_count"] = len(stack_promo_ids) return classified_promotions def _calculate_final_price(self, product_data: Dict, classified_promotions: List[Dict], spec_item_id: Optional[str] = None) -> float: """计算实际到手价：基于促销叠加规则，结合商品价格（支持多规格），计算最低到手价[2]""" # 1. 获取基础价格（多规格则取对应规格价格，否则取商品基础价格） if spec_item_id: # 多规格：获取该规格的价格 spec_price_map = self.parser.adapter.client.get_raw_detail(product_id=product_data["base_info"]["product_id"])["data"].get("specPriceInfo", {}).get("specPriceList", []) spec_price = next((p for p in spec_price_map if p.get("specItemId") == spec_item_id), {}) base_price = spec_price.get("currentPrice", 0.0) else: # 单规格：获取商品基础价格 base_price = product_data.get("price_info", {}).get("current_price", 0.0) if base_price <= 0: return 0.0 final_price = base_price # 2. 按促销优先级（从高到低）应用促销规则 priority_promotions = sorted( classified_promotions, key=lambda x: ["FULL_GIFT", "COUPON", "BRAND_SUBSIDY", "MEMBER_PRICE", "FULL_REDUCE", "LIMITED_DISCOUNT"].index(x["promotion_type"]) ) used_promotions = [] for promo in priority_promotions: if promo["promotion_id"] in used_promotions: continue # 已使用的促销不再重复应用 # 应用促销规则，计算临时价格 temp_price = final_price if promo["promotion_type"] == "LIMITED_DISCOUNT": temp_price = promo["detail"]["discount_price"] elif promo["promotion_type"] == "FULL_REDUCE": temp_price = self._apply_full_reduce(temp_price, promo["detail"]["stages"]) elif promo["promotion_type"] == "COUPON": temp_price = self._apply_coupon(temp_price, promo["detail"]) elif promo["promotion_type"] == "MEMBER_PRICE": temp_price = promo["detail"]["member_price"] elif promo["promotion_type"] == "BRAND_SUBSIDY": temp_price = promo["detail"]["subsidy_price"] elif promo["promotion_type"] == "FULL_GIFT": # 满赠不影响价格，仅标记 used_promotions.append(promo["promotion_id"]) continue # 判断是否可叠加当前促销 if all(p not in self.promotion_stack_rule.get(promo["promotion_type"], []) for p in [u["promotion_type"] for u in used_promotions]): continue # 更新最终价格，标记已使用促销 final_price = temp_price if temp_price < final_price else final_price used_promotions.append(promo["promotion_id"]) # 确保到手价不低于0 return max(final_price, 0.0) def _apply_full_reduce(self, current_price: float, full_reduce_stages: List[Dict]) -> float: """应用满减规则：根据当前价格，匹配最高档位满减""" # 按满减门槛从高到低排序 sorted_stages = sorted(full_reduce_stages, key=lambda x: x["full_amount"], reverse=True) for stage in sorted_stages: if current_price >= stage["full_amount"]: # 计算减免后的价格（不低于0） reduced_price = current_price - stage["reduce_amount"] return max(reduced_price, 0.0) return current_price def _apply_coupon(self, current_price: float, coupon_detail: Dict) -> float: """应用优惠券规则：满足门槛则减免优惠券面额""" if current_price >= coupon_detail["full_amount"]: coupon_price = current_price - coupon_detail["denomination"] return max(coupon_price, 0.0) return current_price def get_promotion_detail(self, product_id: str, region: Optional[str] = None, spec_item_id: Optional[str] = None) -> Dict: """ 全流程促销规则解析：获取规格联动数据→分类促销→判断叠加逻辑→计算到手价 :param product_id: 商品ID :param region: 地区编码 :param spec_item_id: 规格项ID（可选，多规格时需指定） :return: 标准化的促销规则解析数据 """ # 1. 获取规格-价格-库存联动数据 link_detail = self.parser.get_spec_stock_link_detail(product_id=product_id, region=region) if link_detail.get("code") != 0 or not link_detail.get("data"): return link_detail product_data = link_detail["data"] # 仅特卖进行中，解析促销规则（其他阶段无促销） if product_data["sale_time_info"]["sale_stage"] != "ONGOING": product_data["promotion_detail_info"] = {"msg": "特卖未开始/已结束，无促销规则", "data": []} return link_detail # 2. 促销类型分类 classified_promotions = self._classify_promotions(product_data) if not classified_promotions: product_data["promotion_detail_info"] = {"msg": "该商品无有效促销", "data": [], "final_price": product_data["price_info"]["current_price"]} return link_detail # 3. 判断促销叠加逻辑 stacked_promotions = self._judge_stack_logic(classified_promotions) # 4. 计算实际到手价 final_price = self._calculate_final_price(product_data, stacked_promotions, spec_item_id) # 补充促销解析信息 product_data["promotion_detail_info"] = { "msg": "success", "promotion_count": len(stacked_promotions), "promotions": stacked_promotions, "final_price": final_price, "final_price_note