国内主流电商淘宝+京东+1688api数据接口实操-阿里云开发者社区

提供一个获取淘宝、京东和 1688 电商平台 API 数据的 Python 实现方案。这个方案包含了 API 接口的封装、请求处理、数据解析等功能，并提供了完整的异常处理机制。

`ecommerce_api.py
import requests
import time
import hmac
import hashlib
import json
import base64
from typing import Dict, Any, Optional, List

class APIError(Exception):
"""API请求错误"""
def init(self, code: str, message: str, details: Optional[Dict[str, Any]] = None):
self.code = code
self.message = message
self.details = details
super().init(f"APIError: {code} - {message}")

class RateLimitError(APIError):
"""请求频率限制错误"""
def init(self, retry_after: int, message: str = "请求频率超过限制"):
self.retry_after = retry_after
super().init("RATE_LIMIT_EXCEEDED", message, {"retry_after": retry_after})

class AuthError(APIError):
"""认证失败错误"""
def init(self, message: str = "认证失败"):
super().init("AUTHENTICATION_FAILED", message)

class EcommerceAPI:
"""电商API基类"""
def init(self, app_key: str, app_secret: str, timeout: int = 30):
self.app_key = app_key
self.app_secret = app_secret
self.timeout = timeout
self.session = requests.Session()

def _generate_signature(self, params: Dict[str, Any]) -> str:
    """生成API签名（需要根据具体平台的签名算法实现）"""
    raise NotImplementedError("子类必须实现签名生成方法")

def _prepare_params(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
    """准备API请求参数（添加公共参数等）"""
    common_params = {
        "app_key": self.app_key,
        "timestamp": int(time.time() * 1000),
        "format": "json",
        "v": "2.0",
        "method": method
    }
    merged_params = {**common_params, **params}
    merged_params["sign"] = self._generate_signature(merged_params)
    return merged_params

def _send_request(self, url: str, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
    """发送API请求并处理响应"""
    try:
        # 准备请求参数
        prepared_params = self._prepare_params(method, params)

        # 发送请求
        response = self.session.post(url, data=prepared_params, timeout=self.timeout)
        response.raise_for_status()

        # 解析响应
        result = response.json()

        # 检查API返回状态（需要根据具体平台的返回格式调整）
        if "error_response" in result:
            error = result["error_response"]
            error_code = error.get("code", "UNKNOWN_ERROR")
            error_msg = error.get("msg", "未知错误")
            error_sub_code = error.get("sub_code")
            error_sub_msg = error.get("sub_msg")

            if error_code == "isv.access-control-exceed-limit":
                # 处理频率限制错误
                retry_after = int(error_sub_msg.split(" ")[-1]) if error_sub_msg else 60
                raise RateLimitError(retry_after, error_msg)
            elif error_code in ["isv.invalid-parameter:app-key", "isv.invalid-appsecret"]:
                raise AuthError(error_msg)
            else:
                raise APIError(error_code, error_msg, {"sub_code": error_sub_code, "sub_msg": error_sub_msg})

        return result

    except requests.exceptions.Timeout:
        raise APIError("REQUEST_TIMEOUT", "请求超时")
    except requests.exceptions.ConnectionError:
        raise APIError("CONNECTION_ERROR", "连接错误")
    except requests.exceptions.HTTPError as e:
        raise APIError(f"HTTP_ERROR_{e.response.status_code}", f"HTTP错误: {e.response.status_code}")
    except json.JSONDecodeError:
        raise APIError("RESPONSE_PARSE_ERROR", "响应解析错误")
    except RateLimitError:
        raise  # 直接重新抛出频率限制错误
    except Exception as e:
        raise APIError("UNKNOWN_ERROR", f"未知错误: {str(e)}")    ``js

`` alibaba1688_api.py
from .ecommerce_api import EcommerceAPI, APIError, RateLimitError, AuthError
import time
import hashlib
import hmac
import base64
import json

class Alibaba1688API(EcommerceAPI):
"""1688API接口实现"""
BASE_URL = "https://gw.open.1688.com/openapi/param2/1"

def __init__(self, app_key: str, app_secret: str, access_token: Optional[str] = None, timeout: int = 30):
    super().__init__(app_key, app_secret, timeout)
    self.access_token = access_token

def _generate_signature(self, params: Dict[str, Any]) -> str:
    """生成1688API签名"""
    # 1. 按照参数名的字典序升序排列
    sorted_params = sorted(params.items(), key=lambda x: x[0])

    # 2. 拼接参数名和参数值
    string_to_sign = ""
    for key, value in sorted_params:
        string_to_sign += f"{key}{value}"

    # 3. 使用HMAC-SHA1加密
    h = hmac.new(self.app_secret.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha1)
    signature = base64.b64encode(h.digest()).decode()
    return signature

def _prepare_params(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
    """准备API请求参数"""
    method_parts = method.split(".")
    namespace = method_parts[0]
    api_name = ".".join(method_parts[1:])

    common_params = {
        "method": method,
        "appKey": self.app_key,
        "timestamp": int(time.time() * 1000),
        "format": "json",
        "v": "1.0",
        "signMethod": "hmac-sha1"
    }

    if self.access_token:
        common_params["access_token"] = self.access_token

    merged_params = {**common_params, **params}
    merged_params["sign"] = self._generate_signature(merged_params)

    return merged_params

def get_item_info(self, product_id: str) -> Dict[str, Any]:
    """获取商品信息"""
    method = "com.alibaba.product.alibaba.product.get"
    params = {
        "productID": product_id,
        "webSite": "1688"
    }
    return self._send_request(f"{self.BASE_URL}/{self.app_key}/{method}", method, params)

def get_item_search(self, keywords: str, page_no: int = 1, page_size: int = 20) -> Dict[str, Any]:
    """搜索商品"""
    method = "com.alibaba.product.alibaba.product.search"
    params = {
        "keywords": keywords,
        "page": page_no,
        "pageSize": page_size,
        "webSite": "1688"
    }
    return self._send_request(f"{self.BASE_URL}/{self.app_key}/{method}", method, params)

def get_order_list(self, startTime: str, endTime: str, page: int = 1, page_size: int = 20) -> Dict[str, Any]:
    """获取订单列表"""
    method = "com.alibaba.trade.alibaba.trade.getSellerOrderList"
    params = {
        "param1": json.dumps({
            "webSite": "1688",
            "createStartTime": startTime,
            "createEndTime": endTime,
            "currentPage": page,
            "pageSize": page_size
        })
    }
    return self._send_request(f"{self.BASE_URL}/{self.app_key}/{method}", method, params)

jd_api.py
from .ecommerce_api import EcommerceAPI, APIError, RateLimitError, AuthError
import time
import hashlib
import json

class JDApi(EcommerceAPI):
"""京东API接口实现"""
BASE_URL = "https://api.jd.com/routerjson"

def __init__(self, app_key: str, app_secret: str, access_token: Optional[str] = None, timeout: int = 30):
    super().__init__(app_key, app_secret, timeout)
    self.access_token = access_token

def _generate_signature(self, params: Dict[str, Any]) -> str:
    """生成京东API签名"""
    # 1. 按照参数名的字典序升序排列
    sorted_params = sorted(params.items(), key=lambda x: x[0])

    # 2. 拼接参数名和参数值
    string_to_sign = self.app_secret
    for key, value in sorted_params:
        string_to_sign += f"{key}{value}"
    string_to_sign += self.app_secret

    # 3. 使用MD5加密
    signature = hashlib.md5(string_to_sign.encode("utf-8")).hexdigest().upper()
    return signature

def _prepare_params(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
    """准备API请求参数"""
    common_params = {
        "app_key": self.app_key,
        "method": method,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        "format": "json",
        "v": "2.0"
    }

    if self.access_token:
        common_params["access_token"] = self.access_token

    # 处理JSON类型的参数
    json_params = {}
    for key, value in params.items():
        if isinstance(value, (dict, list)):
            json_params[key] = json.dumps(value, ensure_ascii=False)
        else:
            json_params[key] = value

    merged_params = {**common_params, **json_params}
    merged_params["sign"] = self._generate_signature(merged_params)
    return merged_params

def get_item_info(self, sku_ids: str) -> Dict[str, Any]:
    """获取商品信息"""
    method = "jingdong.ware.productInfo.get"
    params = {
        "skuIds": sku_ids,
        "field": "skuId,wareName,jdPrice,marketPrice,brandName,categoryName,wareStatus,imageList"
    }
    return self._send_request(self.BASE_URL, method, params)

def get_item_search(self, keyword: str, page: int = 1, page_size: int = 20) -> Dict[str, Any]:
    """搜索商品"""
    method = "jingdong.union.open.goods.jingfen.query"
    params = {
        "goodsReq": {
            "keyword": keyword,
            "pageIndex": page,
            "pageSize": page_size,
            "sortName": "price",
            "sort": "asc"
        }
    }
    return self._send_request(self.BASE_URL, method, params)

def get_order_list(self, startTime: str, endTime: str, page: int = 1, pageSize: int = 20) -> Dict[str, Any]:
    """获取订单列表"""
    method = "jingdong.union.open.order.row.query"
    params = {
        "orderReq": {
            "startTime": startTime,
            "endTime": endTime,
            "pageIndex": page,
            "pageSize": pageSize,
            "type": 1,  # 1:下单时间，2:完成时间，3:更新时间
            "childUnionId": ""
        }
    }
    return self._send_request(self.BASE_URL, method, params)

main.py
from taobao_api import TaobaoAPI
from jd_api import JDApi
from alibaba1688_api import Alibaba1688API
import time

def main():

# 配置信息（请替换为你自己的AppKey和AppSecret）
taobao_config = {
    "app_key": "YOUR_TAOBAO_APP_KEY",
    "app_secret": "YOUR_TAOBAO_APP_SECRET",
    "session_key": "YOUR_TAOBAO_SESSION_KEY"  # 可选，需要授权的API需要此参数
}

jd_config = {
    "app_key": "YOUR_JD_APP_KEY",
    "app_secret": "YOUR_JD_APP_SECRET",
    "access_token": "YOUR_JD_ACCESS_TOKEN"  # 可选，需要授权的API需要此参数
}

alibaba1688_config = {
    "app_key": "YOUR_1688_APP_KEY",
    "app_secret": "YOUR_1688_APP_SECRET",
    "access_token": "YOUR_1688_ACCESS_TOKEN"  # 可选，需要授权的API需要此参数
}

# 初始化API客户端
taobao_api = TaobaoAPI(**taobao_config)
jd_api = JDApi(**jd_config)
alibaba_api = Alibaba1688API(**alibaba1688_config)

# 示例：搜索商品
search_keyword = "手机"

try:
    print("===== 淘宝商品搜索 =====")
    tb_result = taobao_api.get_item_search(q=search_keyword, page_no=1, page_size=5)
    print(json.dumps(tb_result, ensure_ascii=False, indent=2))

    # 控制请求频率，避免触发限流
    time.sleep(1)

    print("\n===== 京东商品搜索 =====")
    jd_result = jd_api.get_item_search(keyword=search_keyword, page=1, page_size=5)
    print(json.dumps(jd_result, ensure_ascii=False, indent=2))

    time.sleep(1)

    print("\n===== 1688商品搜索 =====")
    alibaba_result = alibaba_api.get_item_search(keywords=search_keyword, page_no=1, page_size=5)
    print(json.dumps(alibaba_result, ensure_ascii=False, indent=2))

except RateLimitError as e:
    print(f"请求频率限制错误: 需要等待 {e.retry_after} 秒后重试")
except AuthError as e:
    print(f"认证失败: {e.message}")
except APIError as e:
    print(f"API请求错误: {e.code} - {e.message}")
except Exception as e:
    print(f"未知错误: {str(e)}")

if name == "main":
main()

taobao_api.py
from .ecommerce_api import EcommerceAPI, APIError, RateLimitError, AuthError
import urllib.parse

class TaobaoAPI(EcommerceAPI):
"""淘宝API接口实现"""
BASE_URL = "https://eco.taobao.com/router/rest"

def __init__(self, app_key: str, app_secret: str, session_key: Optional[str] = None, timeout: int = 30):
    super().__init__(app_key, app_secret, timeout)
    self.session_key = session_key

def _generate_signature(self, params: Dict[str, Any]) -> str:
    """生成淘宝API签名"""
    # 1. 按照参数名的字典序升序排列
    sorted_params = sorted(params.items(), key=lambda x: x[0])

    # 2. 拼接参数名和参数值
    string_to_sign = self.app_secret
    for key, value in sorted_params:
        string_to_sign += f"{key}{value}"
    string_to_sign += self.app_secret

    # 3. 使用MD5加密并转换为大写
    signature = hashlib.md5(string_to_sign.encode("utf-8")).hexdigest().upper()
    return signature

def _prepare_params(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
    """准备API请求参数"""
    common_params = super()._prepare_params(method, params)
    if self.session_key:
        common_params["session"] = self.session_key
    return common_params

def get_item_info(self, num_iid: str) -> Dict[str, Any]:
    """获取商品信息"""
    method = "taobao.item.get"
    params = {
        "fields": "num_iid,title,price,pic_url,detail_url,seller_cids,props_name,item_img,skus",
        "num_iid": num_iid
    }
    return self._send_request(self.BASE_URL, method, params)

def get_item_search(self, q: str, page_no: int = 1, page_size: int = 20) -> Dict[str, Any]:
    """搜索商品"""
    method = "taobao.tbk.dg.material.optional"
    params = {
        "q": q,
        "page_no": page_no,
        "page_size": page_size,
        "adzone_id": "YOUR_ADZONE_ID",  # 需要替换为自己的广告位ID
        "platform": 1  # 1:PC，2:无线
    }
    return self._send_request(self.BASE_URL, method, params)

def get_order_list(self, start_time: str, end_time: str, page_no: int = 1, page_size: int = 20) -> Dict[str, Any]:
    """获取订单列表"""
    method = "taobao.tbk.order.details.get"
    params = {
        "start_time": start_time,
        "end_time": end_time,
        "page_no": page_no,
        "page_size": page_size,
        "query_type": 1,  # 1:按照订单创建时间查询
        "tk_status": 12  # 12:订单结算，13:订单付款，14:订单失效，15:订单成功
    }
    return self._send_request(self.BASE_URL, method, params)

以上代码实现了淘宝、京东和 1688 三个电商平台的 API 接口封装，包含了以下功能：
统一的 API 基类，定义了基本的请求流程和错误处理机制
各平台独立的 API 实现，处理特定的认证和参数格式
常用 API 接口的封装，如商品搜索、商品详情、订单查询等
完整的异常处理机制，包括请求超时、频率限制、认证失败等
使用时，你需要先在各平台申请开发者账号并获取 AppKey 和 AppSecret，然后替换示例代码中的配置信息。注意，部分 API 需要额外的授权才能访问，你需要按照各平台的文档完成授权流程。
实际应用中，你可能需要根据具体业务需求调整参数和处理返回结果。此外，为了避免触发 API 调用频率限制，建议在请求之间添加适当的延迟。

国内主流电商淘宝+京东+1688api数据接口实操

ModelScope模型即服务

热门文章

最新文章

相关电子书

相关实验场景