🤝 京东联盟API（jd.union.open.goods.detail.query）获取佣金与券信息实战（附Python源码）

一、前置条件

1. 注册 京东开放平台→ 创建 京东联盟应用（网站/APP/微信）

2. 京盟后台绑定推广位 PID（格式 unionpid___或 jd_union_pid，推广位ID在京盟后台获取）

3. 应用 → 接口权限 → 申请：

• jd.union.open.goods.detail.query（商品详情+佣金+券）

• jd.union.open.goods.query（关键词搜商品，可选）

4. AppKey / AppSecret 同 JOS 通用

⚠️ 京盟接口不传 access_token（公开选品数据），与商家订单接口不同。

二、接口关键信息

三、Python完整调用 + 佣金/券解析

# jd_union_goods_detail.py
"""
京东联盟 API - 商品详情(佣金+券)
jingdong.union.open.goods.detail.query
依赖: requests  (pip install requests)
"""
import hashlib
import json
import requests
from typing import Dict, List
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

class JdUnionClient:
    GW = "https://api.jd.com/routerjson"

    def __init__(self, app_key: str, app_secret: str):
        self.ak = app_key
        self.as_ = app_secret

    # ───── JOS MD5签名（秒级timestamp）─────
    def _sign(self, params: Dict) -> str:
        filt = sorted((k, v) for k, v in params.items()
                       if v is not None and str(v).strip() != '' and k != 'sign')
        qs = ''.join(f"{k}{v}" for k, v in filt)
        return hashlib.md5(f"{self.as_}{qs}{self.as_}".encode()
                          ).hexdigest().upper()

    def _call(self, method: str, biz: Dict) -> Dict:
        api_p = {
            "app_key": self.ak,
            "method": method,
            "timestamp": str(int(time.time())),   # ← 秒级！
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "360buy_param_json": json.dumps(biz, ensure_ascii=False,
                                             separators=(',', ':'))
        }
        api_p["sign"] = self._sign(api_p)

        r = requests.post(self.GW, data=api_p, timeout=15)
        r.raise_for_status()
        d = r.json()

        # 提取
        resp_key = method.replace(".", "_") + "_response"
        if resp_key not in d:
            for k in d:
                if k.endswith("_response"):
                    resp_key = k
                    break
        data = d.get(resp_key, d)
        if "error_response" in str(data):
            err = d.get(resp_key, {}).get("error_response") or d.get("error_response")
            if err:
                raise Exception(f"JOS Union Err[{err.get('code')}]: "
                                 f"{err.get('zh_desc') or err.get('en_desc')}")
            raise Exception(f"JOS Union error: {d}")
        return data

    # ───── 商品详情(佣金+券) ─────
    def get_goods_detail(self, sku_ids: List[int],
                         fields: str = None,
                         pid: str = None) -> List[Dict]:
        """
        sku_ids: 京东SKU ID列表(最长20)
        pid:     推广位ID(部分字段需传，建议填)
        fields:  推荐全传 ↓
        """
        fields = fields or (
            "skuId,skuName,price,couponList,couponPrice,couponStartTime,"
            "couponEndTime,commissionRate,commission,afterPrice,"
            "picUrl,detailUrl,materialUrl,categoryInfo"
        )
        biz = {"skuIds": sku_ids, "fields": fields}
        if pid:
            biz["pid"] = pid

        resp = self._call("jd.union.open.goods.detail.query", biz)
        # 标准返回: {jd_union_open_goods_detail_query_response:{data:[...]}}
        data_wrapper = resp.get("data") or resp.get("result")
        if isinstance(data_wrapper, list):
            return data_wrapper
        if isinstance(data_wrapper, dict):
            return data_wrapper.get("data") or [data_wrapper]
        return []

    # ───── 扁平化解析 ─────
    def parse_one(self, raw: Dict) -> Dict:
        """返回平铺选品关键字段"""
        coupon_list = raw.get("couponList") or []
        best_coupon = max(coupon_list,
                          key=lambda c: float(c.get("couponPrice") or 0),
                          default={})
        commission_rate_wan = int(raw.get("commissionRate") or 0)  # 万分之
        price = float(raw.get("price") or 0)
        coupon_amt = float(best_coupon.get("couponPrice") or 0)
        est_after = max(price - coupon_amt, 0)
        est_commission = round(est_after * commission_rate_wan / 10000, 2)

        return {
            "sku_id": str(raw.get("skuId")),
            "title": raw.get("skuName"),
            "price": price,
            "coupon_amount": coupon_amt,
            "coupon_start": best_coupon.get("couponStartTime"),
            "coupon_end": best_coupon.get("couponEndTime"),
            "coupon_link": best_coupon.get("link") or best_coupon.get("url"),
            "commission_rate_wan": commission_rate_wan,
            "commission_rate_pct": commission_rate_wan / 100,
            "est_after_coupon": est_after,
            "est_commission": est_commission,
            "pic_url": raw.get("picUrl"),
            "material_url": raw.get("materialUrl"),
            "category": raw.get("categoryInfo")
        }

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# =========================================================
# 使用示例
# =========================================================
if __name__ == "__main__":
    client = JdUnionClient(
        app_key="YOUR_JD_APP_KEY",
        app_secret="YOUR_JD_APP_SECRET"
    )
    PID = "YOUR_UNION_PID"   # ← 京盟后台获取的推广位ID（可选但推荐）

    SKU_IDS = [100012345678, 100087654321]   # ← 替换真实京东SKU ID

    try:
        items = client.get_goods_detail(SKU_IDS, pid=PID)
        for raw in items:
            g = client.parse_one(raw)
            print(f"\n• {g['title'][:30]}")
            print(f"  原价:¥{g['price']}  券¥{g['coupon_amount']}  "
                  f"券后≈¥{g['est_after_coupon']}")
            print(f"  佣金率:{g['commission_rate_pct']:.2f}%  "
                  f"预估佣金:¥{g['est_commission']}")
            if g['coupon_link']:
                print(f"  领券: {g['coupon_link'][:60]}...")
    except Exception as e:
        print("❌", e)
        print("→ 确认: AppKey/Secret正确、京盟应用已备案、skuId有效")

四、关键返回字段说明（选品核心）

est_commission = (price - coupon_amount) * commissionRate / 10000

五、避坑清单

六、面试/方案一句话

京东联盟选品 = 京盟应用备案 → jd.union.open.goods.detail.query(skuIds, fields=couponList+commissionRate+afterPrice)→ 取 couponPrice最高券 + commissionRate(万分之)估算佣金，不传 AccessToken，签名同 JOS MD5（秒级时间戳，AppSecret首尾拼KV_ASCII）。