×

深度分析阿里妈妈API接口,用Python脚本实现

万邦科技Lex 万邦科技Lex 发表于2025-08-20 09:28:19 浏览23 评论0

抢沙发发表评论

        免费测试阿里妈妈API接口

阿里妈妈(Alimama)是阿里巴巴集团旗下的营销服务平台,核心业务包括淘宝联盟、直通车、钻展等,其开放平台提供了丰富的 API 接口,支持推广位管理、商品查询、订单跟踪、佣金结算等功能。以下基于官方 API 进行深度分析,重点实现商品推广信息(含佣金、优惠券)的获取,并提供 Python 调用方案。

### 一、阿里妈妈 API 核心特性分析

#### 1. 接口体系与功能域

阿里妈妈 API(以淘宝联盟为例)按业务分为以下核心域:

-   **商品推广**:商品详情查询、高佣商品推荐;

-   **推广管理**:推广位创建、渠道管理;

-   **订单跟踪**:订单列表查询、订单明细;

-   **优惠券**:优惠券领取、商品优惠券查询。

获取商品推广信息(含佣金、到手价、优惠券)需组合调用**商品详情接口**和**优惠券接口**。

#### 2. 认证与请求规范

-   **认证机制**:采用`appkey + appsecret + 签名`认证,部分接口需用户授权(`session`):

    -   `appkey`/`appsecret`:开发者在阿里妈妈开放平台注册应用后获取;

    -   `sign`:通过参数加密生成的签名(核心安全机制);

    -   `session`:用户授权令牌(部分接口需用户登录态)。

-   **请求格式**:支持 GET/POST,参数需 URL 编码,响应为 JSON/XML 格式(默认 JSON);

-   **网关地址**:统一为。

#### 3. 签名生成规则(核心)

1.  收集所有请求参数(含公共参数和业务参数);

1.  按参数名 ASCII 码升序排序;

1.  拼接为`key=valuekey=value`格式(无分隔符);

1.  首尾拼接`appsecret`,形成`appsecret + 拼接字符串 + appsecret`;

1.  对字符串进行 MD5 加密,转换为大写即为签名(`sign`)。

### 二、Python 脚本实现:阿里妈妈 API 调用框架

以下实现基于淘宝联盟 API,支持商品详情查询(含佣金、优惠券)、推广链接生成,包含签名生成、接口调用及响应解析。

import requests

import json

import time

import hashlib

import logging

from typing import Dict, List, Optional, Tuple

from requests.exceptions import RequestException


# 配置日志

logging.basicConfig(

    level=logging.INFO,

    format="%(asctime)s - %(levelname)s - %(message)s"

)


class AlimamaAPI:

    def __init__(self, appkey: str, appsecret: str, session: str = ""):

        """

        初始化阿里妈妈API客户端

        :param appkey: 应用appkey(从阿里妈妈开放平台获取)

        :param appsecret: 应用appsecret(密钥)

        :param session: 用户授权session(可选,部分接口需要)

        """

        self.appkey = appkey

        self.appsecret = appsecret

        self.session = session

        self.base_url = "https://eco.taobao.com/router/rest"

        self.session = requests.Session()


    def _generate_sign(self, params: Dict) -> str:

        """生成签名(阿里妈妈API签名规则)"""

        # 1. 按参数名ASCII升序排序

        sorted_params = sorted(params.items(), key=lambda x: x[0])

        # 2. 拼接为keyvaluekeyvalue格式

        sign_str = "".join([f"{k}{v}" for k, v in sorted_params])

        # 3. 首尾拼接appsecret并MD5加密

        sign_str = self.appsecret + sign_str + self.appsecret

        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()


    def _get_timestamp(self) -> str:

        """生成时间戳(格式:yyyy-MM-dd HH:mm:ss)"""

        return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())


    def call_api(self, method: str, biz_params: Dict) -> Optional[Dict]:

        """

        通用API调用方法

        :param method: 接口方法名(如taobao.tbk.item.info.get)

        :param biz_params: 业务参数

        :return: 接口响应数据(业务部分)

        """

        # 1. 公共参数

        public_params = {

            "app_key": self.appkey,

            "method": method,

            "timestamp": self._get_timestamp(),

            "format": "json",

            "v": "2.0",

            "sign_method": "md5"

        }

        # 2. 加入用户session(如需要)

        if self.session:

            public_params["session"] = self.session

        # 3. 合并业务参数

        all_params = {**public_params,** biz_params}

        # 4. 生成签名

        sign = self._generate_sign(all_params)

        all_params["sign"] = sign


        # 5. 发送请求

        try:

            response = self.session.get(

                self.base_url,

                params=all_params,

                timeout=10

            )

            response.raise_for_status()

            result = response.json()


            # 6. 处理错误响应

            if "error_response" in result:

                error = result["error_response"]

                logging.error(f"API错误:{error['msg']}(错误码:{error['code']})")

                return None


            # 7. 提取业务响应(格式:{method}_response)

            response_key = method.replace(".", "_") + "_response"

            if response_key not in result:

                logging.error(f"响应格式错误,缺少{response_key}字段")

                return None


            return result[response_key]


        except RequestException as e:

            logging.error(f"请求异常:{str(e)},接口:{method}")

            return None

        except json.JSONDecodeError:

            logging.error(f"响应解析失败:{response.text[:200]}...,接口:{method}")

            return None


    def get_item_info(self, num_iids: str) -> Optional[List[Dict]]:

        """

        获取商品基础信息(含价格、佣金比例)

        :param num_iids: 商品数字ID列表(逗号分隔,如"123456,789012")

        :return: 商品信息列表

        """

        method = "taobao.tbk.item.info.get"

        params = {

            "num_iids": num_iids,

            "fields": "num_iid,title,pict_url,price,zk_final_price,provcity,"

                      "seller_id,volume,nick,commission_rate,commission"

        }

        response = self.call_api(method, params)

        if not response or "results" not in response:

            return None


        # 解析商品列表

        items = response["results"].get("n_tbk_item", [])

        return [

            {

                "num_iid": item.get("num_iid"),

                "title": item.get("title"),

                "pict_url": item.get("pict_url"),

                "price": float(item.get("price", 0)),  # 原价

                "final_price": float(item.get("zk_final_price", 0)),  # 优惠后价格

                "commission_rate": float(item.get("commission_rate", 0)) / 100,  # 佣金比例(%)

                "commission": float(item.get("commission", 0)),  # 佣金金额

                "sales": item.get("volume"),  # 销量

                "seller": item.get("nick")  # 卖家昵称

            } for item in items

        ]


    def get_item_coupon(self, num_iid: str) -> Optional[Dict]:

        """

        获取商品优惠券信息

        :param num_iid: 商品数字ID

        :return: 优惠券信息(无则返回None)

        """

        method = "taobao.tbk.item.coupon.get"

        params = {"num_iid": num_iid}

        response = self.call_api(method, params)

        if not response or "coupons" not in response:

            return None


        coupons = response["coupons"].get("coupon", [])

        if not coupons:

            return None


        # 取第一个有效的优惠券

        coupon = coupons[0]

        return {

            "coupon_id": coupon.get("coupon_id"),

            "total_count": coupon.get("total_count"),  # 总数量

            "remain_count": coupon.get("remain_count"),  # 剩余数量

            "discount": float(coupon.get("discount", 0)),  # 优惠金额

            "min_fee": float(coupon.get("min_fee", 0)),  # 使用门槛

            "start_time": coupon.get("start_time"),

            "end_time": coupon.get("end_time"),

            "coupon_url": coupon.get("coupon_click_url")  # 领取链接

        }


    def generate_promotion_link(self, num_iid: str, adzone_id: str) -> Optional[Dict]:

        """

        生成商品推广链接(含优惠券)

        :param num_iid: 商品数字ID

        :param adzone_id: 推广位ID(需提前创建)

        :return: 推广链接信息

        """

        method = "taobao.tbk.promotion.link.generate"

        params = {

            "material_id": num_iid,  # 商品ID

            "adzone_id": adzone_id,

            "material_type": "item"  # 类型为商品

        }

        response = self.call_api(method, params)

        if not response or "result_list" not in response:

            return None


        result = response["result_list"].get("map_data", [{}])[0]

        return {

            "click_url": result.get("click_url"),  # 推广链接

            "coupon_click_url": result.get("coupon_click_url"),  # 带优惠券的推广链接

            "short_url": result.get("short_url")  # 短链接

        }


    def get_item_promotion_info(self, num_iid: str, adzone_id: str = "") -> Optional[Dict]:

        """

        获取商品完整推广信息(基础信息+优惠券+推广链接)

        :param num_iid: 商品数字ID

        :param adzone_id: 推广位ID(可选,用于生成推广链接)

        :return: 完整推广信息

        """

        # 1. 获取商品基础信息

        item_info = self.get_item_info(num_iid)

        if not item_info or len(item_info) == 0:

            return None

        item = item_info[0]


        # 2. 获取优惠券信息

        coupon = self.get_item_coupon(num_iid)


        # 3. 生成推广链接(如有推广位ID)

        promotion_link = None

        if adzone_id:

            promotion_link = self.generate_promotion_link(num_iid, adzone_id)


        # 4. 计算最终推广价(优惠后价格-优惠券,如有)

        final_promotion_price = item["final_price"]

        if coupon and final_promotion_price >= coupon["min_fee"]:

            final_promotion_price -= coupon["discount"]


        return {

            "item": item,

            "coupon": coupon,

            "promotion_link": promotion_link,

            "final_promotion_price": round(final_promotion_price, 2),  # 最终推广价

            "estimated_commission": round(final_promotion_price * item["commission_rate"], 2)  # 预估佣金

        }



# 示例调用

if __name__ == "__main__":

    # 替换为实际参数(从阿里妈妈开放平台获取)

    APPKEY = "your_appkey"

    APPSECRET = "your_appsecret"

    SESSION = "your_session"  # 可选,部分接口需要

    ADZONE_ID = "your_adzone_id"  # 推广位ID(需提前创建)


    # 初始化客户端

    alimama = AlimamaAPI(

        appkey=APPKEY,

        appsecret=APPSECRET,

        session=SESSION

    )


    # 示例商品ID(淘宝商品数字ID)

    ITEM_ID = "6543217890"


    # 获取商品完整推广信息

    promotion_info = alimama.get_item_promotion_info(ITEM_ID, ADZONE_ID)


    if promotion_info:

        item = promotion_info["item"]

        print(f"商品标题:{item['title']}")

        print(f"原价:{item['price']}元 | 优惠后价格:{item['final_price']}元")

        print(f"佣金比例:{item['commission_rate'] * 100}% | 预估佣金:{promotion_info['estimated_commission']}元")

        

        coupon = promotion_info["coupon"]

        if coupon:

            print(f"优惠券:满{coupon['min_fee']}元减{coupon['discount']}元(剩余{coupon['remain_count']}/{coupon['total_count']})")

            print(f"优惠券有效期:{coupon['start_time']} 至 {coupon['end_time']}")

        

        print(f"最终推广价:{promotion_info['final_promotion_price']}元")

        

        link = promotion_info["promotion_link"]

        if link:

            print(f"推广链接:{link['click_url']}")

            if link["coupon_click_url"]:

                print(f"带券推广链接:{link['coupon_click_url']}")

### 三、关键技术点解析

#### 1. 核心接口解析

-   **`taobao.tbk.item.info.get`**:获取商品基础信息,包括原价(`price`)、优惠后价格(`zk_final_price`)、佣金比例(`commission_rate`)等,是计算推广收益的基础;

-   **`taobao.tbk.item.coupon.get`**:查询商品可用优惠券,返回优惠金额(`discount`)、使用门槛(`min_fee`)等,用于计算最终到手价;

-   **`taobao.tbk.promotion.link.generate`**:生成带推广位的商品链接,用于跟踪推广效果和佣金归属,需传入`adzone_id`(推广位 ID)。

#### 2. 价格与佣金计算逻辑

-   **最终推广价** = 优惠后价格(`zk_final_price`) - 优惠券金额(`discount`,需满足使用门槛`min_fee`);

-   **预估佣金** = 最终推广价 × 佣金比例(`commission_rate`,需转换为小数,如 15% → 0.15)。

#### 3. 推广位管理

`adzone_id`(推广位 ID)是跟踪推广效果的关键,需通过接口创建,关联到具体渠道(`site_id`),创建后可在阿里妈妈后台查看推广数据。

### 四、使用注意事项

1.  **权限申请**:

    -   需在注册开发者,创建应用并绑定淘宝联盟账号;

    -   部分高权限接口(如高佣商品查询)需通过平台审核,个人开发者需完成实名认证。

1.  **限流控制**:

    -   普通接口 QPS 限制为 10,超过会返回`429 Too Many Requests`;

    -   建议添加请求间隔(如 1 秒 / 次),批量查询时使用多线程控制频率。

1.  **参数规范**:

    -   `num_iid`(商品 ID)需从淘宝商品详情页 URL 提取;

    -   时间戳需与阿里妈妈服务器时间同步(误差≤10 分钟),否则签名验证失败。

1.  **合规性**:

    -   推广链接需明确标注 “广告” 或 “推广”,遵守《电子商务法》;

    -   禁止通过 API 获取数据用于不正当竞争或侵权用途。

该实现基于官方 API,适用于电商导购平台、推广工具等场景,通过扩展可支持批量商品查询、订单跟踪等功能。实际使用需参考调整参数和接口逻辑。


群贤毕至

访客