
|
参数名 |
类型 |
说明 |
是否必选 |
技术约束 |
|
app_key |
String |
应用唯一标识(开放平台申请) |
是 |
长度固定 32 位,仅含字母和数字,不可包含特殊字符 |
|
app_secret |
String |
接口签名密钥 |
是 |
需通过环境变量读取,禁止硬编码至代码(避免泄露) |
|
method |
String |
接口名称 |
是 |
固定为 “jd.ware.get”(注意区分 “ware” 与 “item”,写错会报接口不存在) |
|
ware_id |
String |
商品唯一标识(京东标准 SKU ID) |
是 |
纯数字格式,长度 10-15 位,需从商品详情页 URL 或列表接口获取 |
|
timestamp |
String |
请求时间戳 |
是 |
格式严格为 “YYYY-MM-DD HH:MM:SS”,与京东服务器时间偏差≤3 分钟(超差报签名错) |
|
format |
String |
响应格式 |
是 |
仅支持 “json”,不支持 xml(传 xml 会返回格式错误) |
|
v |
String |
接口版本 |
是 |
稳定版本为 “2.0”,低版本(1.0)已停用,调用会返回 “接口已下线” |
|
sign_method |
String |
签名算法 |
是 |
固定为 “md5”,无其他可选算法(传 sha1 会报签名方式错误) |
|
fields |
String |
返回字段列表 |
否 |
建议按需选择(如 “ware_id,title,price,stock”),全字段请求会增加响应耗时 |
import hashlibimport sortedcontainersfrom typing import Dictdef generate_jd_sign(params: Dict, app_secret: str) -> str:"""生成京东接口签名(严格遵循MD5算法规范,解决90%签名错误):param params: 待签名参数字典(不含sign字段):param app_secret: 应用密钥:return: 32位大写签名字符串"""# 避坑点1:必须按参数名ASCII码升序排序(普通dict无序,需用SortedDict)sorted_params = sortedcontainers.SortedDict(params)# 避坑点2:拼接格式为"keyvalue"(无分隔符),首尾需加app_secretsign_str = app_secretfor key, value in sorted_params.items():# 避坑点3:参数值需转为字符串(数字/布尔类型直接拼接会导致签名偏差)sign_str += f"{key}{str(value)}"sign_str += app_secret# 避坑点4:需用UTF-8编码(中文参数如商品标题不编码会导致加密错误)return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()import requestsimport jsonimport timeimport loggingfrom dataclasses import dataclassfrom typing import Optional, List# 日志配置(技术调试必备,记录错误详情便于排查)logging.basicConfig(level=logging.INFO,format='%(asctime)s - jd-ware-api - %(levelname)s - %(message)s')logger = logging.getLogger("jd-ware-client")# 商品详情数据模型(结构化存储,避免数据混乱)@dataclassclass JdWareDetail:ware_id: str # 商品IDtitle: str # 商品标题price: float # 商品售价stock: int # 库存数量sales: int # 累计销量category_id: str # 类目IDshop_id: str # 店铺IDspec_list: List[dict] # 规格列表(如颜色、尺寸)create_time: str # 上架时间service_tags: List[str] # 服务标签(如“7天无理由”)class JdWareAPIClient:def __init__(self, app_key: str, app_secret: str, timeout: tuple = (3, 10)):self.app_key = app_keyself.app_secret = app_secretself.base_url = "https://api.jd.com/routerjson" # 京东接口固定地址self.timeout = timeout# 初始化长连接(减少TCP握手开销,提升并发性能)self.session = self._init_session()def _init_session(self) -> requests.Session:"""初始化请求会话,配置连接池与重试机制"""session = requests.Session()adapter = requests.adapters.HTTPAdapter(pool_connections=10, # 连接池数量pool_maxsize=50, # 每个连接池最大连接数# 重试策略:针对5xx错误自动重试3次(避免临时网络问题)max_retries=requests.packages.urllib3.util.retry.Retry(total=3,status_forcelist=[500, 502, 503, 504],backoff_factor=0.5 # 重试间隔:0.5s, 1s, 2s))session.mount('https://', adapter)return sessiondef _validate_params(self, params: Dict) -> bool:"""参数校验(避免因格式错误导致请求失败)"""# 校验ware_id格式(纯数字)if not params.get("ware_id", "").isdigit():logger.error(f"ware_id格式错误:{params.get('ware_id')}(需纯数字)")return False# 校验时间戳格式try:time.strptime(params.get("timestamp"), "%Y-%m-%d %H:%M:%S")except ValueError:logger.error(f"timestamp格式错误:{params.get('timestamp')}(需YYYY-MM-DD HH:MM:SS)")return False# 校验fields字段(避免空值或无效字段)fields = params.get("fields", "")if fields and " " in fields: # 字段间用逗号分隔,不可有空格logger.error(f"fields格式错误:{fields}(字段间不可有空格)")return Falsereturn Truedef get_ware_detail(self, ware_id: str, fields: str = "ware_id,title,price,stock,sales") -> Optional[JdWareDetail]:"""核心方法:获取商品详情数据:param ware_id: 商品ID:param fields: 返回字段列表(逗号分隔):return: 结构化JdWareDetail对象(失败返回None)"""# 1. 构建基础参数字典base_params = {"method": "jd.ware.get","app_key": self.app_key,"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),"format": "json","v": "2.0","sign_method": "md5","ware_id": ware_id,"fields": fields}# 2. 参数校验与签名生成if not self._validate_params(base_params):return Nonebase_params["sign"] = generate_jd_sign(base_params, self.app_secret)# 3. 发送请求(京东接口仅支持POST,GET会返回405 Method Not Allowed)try:response = self.session.post(url=self.base_url,data=base_params, # 参数放入data字段(非params)headers={"Content-Type": "application/x-www-form-urlencoded"},timeout=self.timeout)response.raise_for_status() # 捕获4xx/5xx HTTP错误(如403权限不足)# 4. 解析响应数据try:result = json.loads(response.text)except json.JSONDecodeError as e:logger.error(f"JSON解析失败:{str(e)},响应内容:{response.text[:500]}")return None# 5. 处理业务错误(京东错误码规范:0为成功)if result.get("code") != 0:error_msg = result.get("message", "未知业务错误")error_code = result.get("code", "未知错误码")logger.error(f"接口业务错误:{error_msg}(错误码:{error_code})")return None# 6. 结构化商品数据(处理字段缺失与格式转换)raw_data = result.get("result", {}).get("wareInfo", {})# 处理规格列表(可能返回空列表,需兼容)spec_list = raw_data.get("specList", [])# 处理销量字段(部分商品返回字符串,需转为int)sales = int(raw_data.get("sales", 0)) if raw_data.get("sales").isdigit() else 0return JdWareDetail(ware_id=raw_data.get("ware_id", ""),title=raw_data.get("title", ""),price=float(raw_data.get("price", 0.0)),stock=int(raw_data.get("stock", 0)),sales=sales,category_id=raw_data.get("categoryId", ""),shop_id=raw_data.get("shopId", ""),spec_list=spec_list,create_time=raw_data.get("createTime", ""),service_tags=raw_data.get("serviceTags", []))except requests.exceptions.RequestException as e:logger.error(f"请求异常:{str(e)}")return Noneexcept Exception as e:logger.error(f"未知处理异常:{str(e)}")return Nonedef clean_ware_data(ware_detail: JdWareDetail) -> Optional[JdWareDetail]:"""商品数据清洗:处理异常值与格式统一,为商业分析做准备:param ware_detail: 原始结构化数据:return: 清洗后的数据(异常数据返回None)"""# 1. 过滤无效商品(标题为空或ware_id无效)if not ware_detail.title.strip() or not ware_detail.ware_id.isdigit():logger.warning(f"无效商品:ware_id={ware_detail.ware_id},标题为空")return None# 2. 处理价格异常(如0元、负数,设为默认值)if ware_detail.price <= 0:logger.warning(f"价格异常:ware_id={ware_detail.ware_id},价格={ware_detail.price},重置为999.99")ware_detail.price = 999.99# 3. 处理库存异常(负数库存设为0,超10万库存设为10万(避免异常值影响分析))if ware_detail.stock < 0:ware_detail.stock = 0elif ware_detail.stock > 100000:ware_detail.stock = 100000# 4. 规格列表标准化(提取“规格名-规格值”键值对,便于后续筛选)standardized_spec = []for spec in ware_detail.spec_list:spec_name = spec.get("specName", "").strip()spec_value = spec.get("specValue", "").strip()if spec_name and spec_value:standardized_spec.append({"name": spec_name, "value": spec_value})ware_detail.spec_list = standardized_specreturn ware_detaildef calculate_stock_days(ware_detail: JdWareDetail, avg_daily_sales: int) -> float:"""计算库存可售天数(支撑库存预警决策):param ware_detail: 商品详情数据:param avg_daily_sales: 日均销量(从历史数据统计):return: 库存可售天数(保留1位小数)"""if avg_daily_sales <= 0:return float("inf") # 无销量时,可售天数设为无穷大return round(ware_detail.stock / avg_daily_sales, 1)# 库存预警逻辑if __name__ == "__main__":# 初始化客户端(替换为实际app_key与app_secret)client = JdWareAPIClient(app_key="your_app_key",app_secret="your_app_secret")# 获取商品详情ware_detail = client.get_ware_detail(ware_id="100012014970")if ware_detail:# 数据清洗cleaned_ware = clean_ware_data(ware_detail)if cleaned_ware:# 假设日均销量为50(实际从历史数据统计)stock_days = calculate_stock_days(cleaned_ware, avg_daily_sales=50)# 库存预警(可售天数≤3天)if stock_days <= 3:logger.warning(f"【库存预警】商品{cleaned_ware.title}(ID:{cleaned_ware.ware_id})可售天数仅{stock_days}天,需补货")else:logger.info(f"【库存正常】商品{cleaned_ware.title}(ID:{cleaned_ware.ware_id})可售天数{stock_days}天")def analyze_price_competitiveness(target_ware: JdWareDetail, competitor_wares: List[JdWareDetail]) -> dict:"""价格竞争力分析:计算价格偏离度,支撑定价决策:param target_ware: 目标商品:param competitor_wares: 竞品商品列表:return: 价格分析结果(含均价、偏离度)"""# 过滤无效竞品(价格异常或与目标商品类目不符)valid_competitors = [w for w in competitor_waresif w.price > 0 and w.category_id == target_ware.category_id]if not valid_competitors:return {"status": "fail", "msg": "无有效竞品数据"}# 计算竞品均价avg_competitor_price = sum(w.price for w in valid_competitors) / len(valid_competitors)# 计算价格偏离度(正值:目标商品价格高于竞品,负值:低于竞品)price_deviation = round(((target_ware.price - avg_competitor_price) / avg_competitor_price) * 100, 2)# 价格竞争力判断if price_deviation < -10:competitiveness = "强(价格低于竞品10%以上)"elif -10 <= price_deviation <= 10:competitiveness = "中等(价格与竞品持平)"else:competitiveness = "弱(价格高于竞品10%以上)"return {"status": "success","target_ware_id": target_ware.ware_id,"target_price": target_ware.price,"competitor_avg_price": round(avg_competitor_price, 2),"price_deviation": f"{price_deviation}%","competitiveness": competitiveness}# 价格分析示例if __name__ == "__main__":# 假设已获取目标商品与3个竞品的详情数据target_ware = JdWareDetail(ware_id="100012014970", title="某品牌手机", price=2999.0, stock=1000, sales=5000,category_id="1001", shop_id="12345", spec_list=[], create_time="2024-01-01", service_tags=[])competitor_wares = [JdWareDetail(ware_id="100012014971", title="竞品1", price=2899.0, stock=800, sales=4500, category_id="1001", shop_id="67890", spec_list=[], create_time="2024-01-02", service_tags=[]),JdWareDetail(ware_id="100012014972", title="竞品2", price=2999.0, stock=1200, sales=5200, category_id="1001", shop_id="13579", spec_list=[], create_time="2024-01-03", service_tags=[]),JdWareDetail(ware_id="100012014973", title="竞品3", price=3099.0, stock=600, sales=3800, category_id="1001", shop_id="24680", spec_list=[], create_time="2024-01-04", service_tags=[])]# 执行价格分析price_analysis = analyze_price_competitiveness(target_ware, competitor_wares)if price_analysis["status"] == "success":logger.info(f"价格分析结果:{json.dumps(price_analysis, ensure_ascii=False, indent=2)}")|
技术问题 |
错误表现 |
解决方案(技术细节) |
|
签名无效(错误码 10003) |
接口返回 “签名无效”,请求被拒绝 |
1. 用 SortedDict 确保参数 ASCII 升序;2. 检查参数值是否均转为字符串;3. 验证 timestamp 与服务器时间差≤3 分钟;4. 确认 app_secret 与应用匹配 |
|
请求方式错误(405) |
报 “Method Not Allowed”,请求失败 |
强制使用 POST 方法,参数放入 data 字段(而非 params),设置 Content-Type 为 “application/x-www-form-urlencoded” |
|
数据解析异常 |
JSONDecodeError,或 specList 字段格式混乱 |
1. 捕获 JSON 解析异常并打印响应片段;2. 对 specList 做空列表判断;3. 销量字段先校验是否为数字再转 int |
|
权限不足(错误码 10004) |
接口返回 “您没有该接口的调用权限” |
1. 检查应用是否已申请 “jd.ware.get” 接口权限;2. 确认开发者认证已通过;3. 企业开发者需确保营业执照在有效期内 |
|
调用超限(错误码 10005) |
接口返回 “调用频率超限” |
1. 实现令牌桶算法控制 QPS(个人≤1,企业≤3);2. 失败后延迟 3-5 秒重试;3. 按 “ware_id + 时间戳” 记录请求进度,避免重复调用 |
|
商品不存在(错误码 2001) |
接口返回 “商品不存在或已下架” |
1. 验证 ware_id 是否正确(可在京东 APP 搜索确认);2. 对已下架商品做标记,避免重复请求;3. 从列表接口重新获取有效 ware_id |
需要获取更多接口测试,欢迎联系小编必回
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/70047598/viewspace-3097218/ ,如需转载,请注明出处,否则将追究法律责任。